/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2020-2023. All rights reserved.
 * Description: hccl实现基类文件
 */

#include <atomic>
#include <chrono>
#include <thread>
#include <algorithm>
#include <numeric>
#include <unordered_set>
#include <sys/time.h>
#include "slog.h"
#include "slog_api.h"
#include "adapter/adapter_hal.h"
#include "adapter/adapter_rts.h"
#include "adapter/adapter_hdds.h"
#include "adapter_prof.h"
#include "dlprof_function.h"
#include "dlrt_function.h"
#include "externalinput.h"
#include "p2p_mgmt_pub.h"
#include "../command_handle.h"
#include "config.h"
#include "transport_heterog_event_roce_pub.h"
#include "transport_heterog_event_tcp_pub.h"
#include "transport_heterog_roce_pub.h"
#include "transport_roce.h"
#include "alg_coll_remote_update.h"
#include "callback_thread_manager.h"
#include "transport_shm_event_pub.h"
#include "stream_active_manager.h"
#include "device_capacity.h"
#include "profiling_manager.h"
#include "task_exception_handler_pub.h"
#include "hccl_impl_base.h"
#include "remote_notify.h"
#include "rank_consistentcy_checker.h"

using namespace std;

namespace hccl {
static std::mutex g_hcomInitMutex;
constexpr u32 MEMORY_CAPACITY = 256 * 1024;
constexpr u32 WAIT_PREPARE_SLEEP_TIME = 5000;
constexpr u32 SINGLE_SERVER_NUM = 1;
constexpr u32 CONN_LIMIT = 4096;
constexpr u32 ES_STREAM_LIMIT = 8;
constexpr u32 OP_NUM = 2;

enum TransferMemInfoIdx {
    TRANSFER_MEM_INFO_KEY_IDX = 0,
    TRANSFER_MEM_INFO_VALUE_IDX = 1,
    TRANSFER_MEM_INFO_RDMA_ENVELOPE_IDX = 2,
    TRANSFER_MEM_INFO_IDX_NUM = 3
};

constexpr uint32_t AUX_EVENT_ID = 51U;
constexpr uint32_t ORIGINAL_THREAD_ID = 512U;
constexpr u32 MIN_QP_DEPTH = 8;
constexpr u32 MAX_QP_DEPTH = 512;
constexpr u32 MAX_QP_DEPTH_HCCP = 4096;
constexpr u32 DEVICE_LOGIC_ID_ZERO = 0;

HcclImplBase::HcclImplBase()
    : dispatcher_(nullptr), vDispatcher_(nullptr), notifyPool_(nullptr), queueNotifyManager_(nullptr),
      initializedFlag_(ATOMIC_FLAG_INIT), userRank_(INVALID_VALUE_RANKID), realUserRank_(INVALID_VALUE_RANKID),
      userRankSize_(INVALID_VALUE_RANKSIZE), drvInit_(false), inlineReduceSwitchOn_(true),
      nicDeployment_(NICDeployment::NIC_DEPLOYMENT_DEVICE), devicePhyId_(INVALID_UINT),
      deviceLogicId_(-1), localRank_(INVALID_VALUE_RANKID), hostSocketHandle_(nullptr),
      isUsedRdmaLevel0_(false), nicInitialized_(false), heterogHostNicInitialized_(false), hcomGroupNicInit_(false),
      profilingMode_(HcomProfilingMode::PROFILING_CLOSE), raResourceInit_(false),
      interServer_(false), isSingleMeshAggregation_(false), cclBufferManager_(CCLBufferManager()),
      isExecuteProfilingInit_(false), deviceType_(DevType::DEV_TYPE_COUNT),
      heterogRaInit_(false), isHeterogComm_(false), hostRdmaInitFlag_(false),
      csCommInitFlag_(false), commHandle_(nullptr),
      commWorkMode_(WorkMode::HCCL_MODE_NORMAL), meshAggregationRankSize_(0), isHaveCpuRank_(false), ranktableCrc_(0),
      pMsgInfosMem_(nullptr), pReqInfosMem_(nullptr), memBlocksManager_(nullptr), pRecvWrInfosMem_(nullptr),
      transportResourceInfo_(mrManager_, pMsgInfosMem_, pReqInfosMem_, memBlocksManager_, pRecvWrInfosMem_),
      multiModuleDiffDeviceNumMode_(false), isStandardCard_(false),
      loopBackIp_(HcclIpAddress(LOOPBACK_IP)), profilingInitiated_(false), callbackThreadId_(INVALID_U64),
      role_(SERVER_ROLE_SOCKET), psRaInit_(false), workerRaInit_(false), mrManagerInit_(false), srqInit_(false),
      firstBuildTransportFlag_(true), isHostUseDevNic_(false),
      buildTransportRefCount_(0), embeddingReduceThreadNum_(REDUCE_THREAD_NUM), isCpuRank_(false),
      isAllRankSamePlane_(false), serverNum_(0), moduleNum_(0), threadPrepareCommAsyncExit_(false),
      opBaseAtraceInfo_(nullptr)
{
    mrManager_.reset(new (std::nothrow) MrManager());
    if (mrManager_ == nullptr) {
        HCCL_ERROR("new MrManager failed!");
    }
}

HcclResult HcclImplBase::DeinitAndClearTransportMap()
{
    // 执行捕获异常
    auto transportHeterogFree = [this] (map<u32, unique_ptr<TransportHeterog>> &mymap) -> void {
        for (auto &transportIt : mymap) {
            if (transportIt.second == nullptr) {
                continue;
            }
            HcclResult ret = transportIt.second->Deinit();
            if (ret != HCCL_SUCCESS) {
                HCCL_ERROR("transport rankId[%u] deinit failed", transportIt.first);
                throw runtime_error("transport deinit failed");
            }
        }

        HCCL_INFO("[HcclImplBase]DestroyCDomainResource transportHeterogMap_ erased by tag");
    };

    EXECEPTION_CATCH(transportHeterogMap_.EraseAll(transportHeterogFree), return HCCL_E_INTERNAL);

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::ClearPsTagNodesInfo()
{
    std::unique_lock<std::mutex> tagNodesInfoLock(tagNodesInfoLock_);
    tagNodesInfo_.clear();
    tagNodesInfoLock.unlock();
    isEsAllLinkModel_ = false;
    HCCL_INFO("ClearPsTagNodesInfo success");
    return HCCL_SUCCESS;
}

HcclImplBase::~HcclImplBase()
{
    if (implAlg_ != nullptr) {
        delete implAlg_;
        implAlg_ = nullptr;
    }

    tagWorkSpaceMem_.clear();

    StopPrepareComm();
    tagOpExecutorMap_.Clear();

    {
        lock_guard<mutex> lock(tagServiceHandleMapMutex_);
        tagServiceHandleMap_.clear();
    }

    CsDeInit();

    if (isHaveCpuRank_ && isSetHDCModeInfo_ && static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
        DeinitHdcPsRaResource();
    }
    (void)UnRegistTaskExceptionHandler();

    if (GetExternalInputHcclIsTcpMode()) {
        TcpSendThreadPool::GetSendPoolInstance()->Deinit();
    }

    // 销毁异构通信资源
    DestroyHeterogTransport();
    DestroySrq();
    DeInitTransportMem();
    MrManagerDeInit();

    /* 网络资源销毁 */
    DestroyNetworkResources();
    DeinitHeterogRaResource();

    /* driver关联资源释放 */
    if (drvInit_) {
        if (DisablePreResource() != HCCL_SUCCESS) {
            HCCL_WARNING("driver resource is not released successfully");
        }
    }

    if (isExecuteProfilingInit_) {
        (void)DeinitProfiling();
    }

    if (!isHeterogComm_) {
        /* 销毁当前trace句柄 */
        if (opBaseAtraceInfo_ != nullptr) {
            opBaseAtraceInfo_->DeInit();
            opBaseAtraceInfo_ = nullptr;
        }
    }

    ReleaseWorkSpacebuffer();

    if (dispatcher_ != nullptr) {
        HcclDispatcherDestroy(dispatcher_);
        dispatcher_ = nullptr;
    }
    if (vDispatcher_ != nullptr) {
        HcclDispatcherDestroy(vDispatcher_);
        vDispatcher_ = nullptr;
    }
}

HcclResult HcclImplBase::ClearOpResource(const std::string &tag)
{
    HCCL_INFO("ClearOpResource tag[%s]", tag.c_str());
    tagCommInfo_.erase(tag);
    // stream解绑定
    auto iterStream = tagStreamInfo_.find(tag);
    if (iterStream != tagStreamInfo_.end()) {
        CHK_RET(StreamActiveManager::GetInstance(deviceLogicId_).StreamsUnactive(iterStream->second.ringStreams));
    }
    tagStreamInfo_.erase(tag);

    if (implAlg_ != nullptr) {
        implAlg_->ClearOpResource(tag);
    }
    DestroyWorkspaceResource(tag);

    unique_lock<std::mutex> lockTransferMemsMap(transferMemsMapMutex_);
    s32 intTag = atoi(tag.c_str());
    transferMemsMap_.erase(intTag);
    lockTransferMemsMap.unlock();

    CHK_RET(DestroyCDomainResource(intTag));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CreateOpBasedResources(const HcclCMDType &opType, const std::string &tag,
    const HcomCollOpInfo &opInfo)
{
    return workSpaceRes_->CreateOpBasedResources(opType, tag, opInfo);
}

HcclResult HcclImplBase::CreateRemoteOpBasedResources(u64 memSize, const std::string &tag)
{
    return workSpaceRes_->CreateRemoteOpBasedResources(memSize, tag);
}

HcclResult HcclImplBase::CreateOrUpdateRemoteOpBasedResources(u64 memSize, const std::string &tag)
{
    return workSpaceRes_->CreateOrUpdateRemoteOpBasedResources(memSize, tag);
}

HcclResult HcclImplBase::DestroyRemoteOpBasedMem(const std::string &tag)
{
    return workSpaceRes_->DestroyRemoteOpBasedMem(tag);
}

bool HcclImplBase::IsAtomicInit()
{
    if (!initializedFlag_.test_and_set()) {
        initializedFlag_.clear();
        return false;
    }
    return true;
}

bool HcclImplBase::IsNeedNicInit()
{
    return ((!nicInitialized_) && (!hcomGroupNicInit_) && (userRankSize_ > 1) && !isSingleMeshAggregation_);
}

HcclResult HcclImplBase::RegisterMemory(void* buffer, uint64_t size)
{
    // 拉远、推理场景PS侧支持注册全局内存
    if (isHeterogComm_) {
        if (hostRdmaInitFlag_) {
            CHK_RET(mrManager_->RegGlobalMr(buffer, size));
        }
    } else {
        HCCL_ERROR("[RegisterMemory]RegisterMemory is not supported.");
        return HCCL_E_NOT_SUPPORT;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::UnregisterMemory(void* buffer)
{
    if (isHeterogComm_) {
        if (hostRdmaInitFlag_) {
            CHK_RET(mrManager_->DeRegGlobalMr(buffer));
        }
    } else {
        HCCL_ERROR("[UnregisterMemory]UnregisterMemory is not supported.");
        return HCCL_E_NOT_SUPPORT;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetBandWidthPerNPU(u32 level, float &bandWidth)
{
    return hccl::GetBandWidthPerNPU(level, userRankSize_, deviceNumPerAggregation_, bandWidth);
}

HcclResult HcclImplBase::GetDeviceNumPerAggregation(u32 &deviceNumPerAggregation)
{
    deviceNumPerAggregation = deviceNumPerAggregation_;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckDataType(const HcclDataType dataType, bool needReduce)
{
    if (needReduce) {
        if ((dataType == HCCL_DATA_TYPE_UINT64) ||
            (dataType == HCCL_DATA_TYPE_UINT8) || (dataType == HCCL_DATA_TYPE_UINT16) ||
            (dataType == HCCL_DATA_TYPE_UINT32) || (dataType == HCCL_DATA_TYPE_FP64) ||
            (dataType == HCCL_DATA_TYPE_RESERVED)) {
            HCCL_ERROR("[Check][DataType]errNo[0x%016llx] data type[%s] not supported",
                HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), GetDataTypeEnumStr(dataType).c_str());
            return HCCL_E_NOT_SUPPORT;
        }
    } else {
        if ((dataType >= HCCL_DATA_TYPE_RESERVED) || (dataType < HCCL_DATA_TYPE_INT8)) {
            HCCL_ERROR("[Check][DataType]errNo[0x%016llx] data type[%s] not supported",
                HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), GetDataTypeEnumStr(dataType).c_str());
            return HCCL_E_NOT_SUPPORT;
        }
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckReduceDataType(const HcclDataType dataType, const HcclReduceOp op)
{
    if ((deviceType_ == DevType::DEV_TYPE_910B) || (deviceType_ == DevType::DEV_TYPE_910_93)) {
        if ((op == HCCL_REDUCE_PROD) &&
        ((dataType == HCCL_DATA_TYPE_INT16) || (dataType == HCCL_DATA_TYPE_BFP16))) {
            HCCL_ERROR(
                "[Check][DataType]errNo[0x%016llx] device type[%d] does not support the data type[%s] and data "\
                "type[%s] for Op[%s]", HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), deviceType_,
                GetDataTypeEnumStr(HcclDataType::HCCL_DATA_TYPE_BFP16).c_str(),
                GetDataTypeEnumStr(HcclDataType::HCCL_DATA_TYPE_INT16).c_str(),
                GetReduceOpEnumStr(op).c_str());
            return HCCL_E_NOT_SUPPORT;
        }
    } else if (deviceType_ == DevType::DEV_TYPE_910) {
        if (dataType == HCCL_DATA_TYPE_INT16) {
            HCCL_ERROR(
                "[Check][DataType]errNo[0x%016llx] device type[%d] does not support the data type[%s]",\
                HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), deviceType_,
                GetDataTypeEnumStr(dataType).c_str());
            return HCCL_E_NOT_SUPPORT;
        }
    } else if (deviceType_ == DevType::DEV_TYPE_310P3) {
        if (dataType == HcclDataType::HCCL_DATA_TYPE_INT16 && op != HcclReduceOp::HCCL_REDUCE_SUM) {
            HCCL_ERROR(
                "[Check][DataType]errNo[0x%016llx] device type[%d] does not support the data type[%s] for Op[%s]",\
                HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), deviceType_,
                GetDataTypeEnumStr(HcclDataType::HCCL_DATA_TYPE_INT16).c_str(),
                GetReduceOpEnumStr(op).c_str());
            return HCCL_E_NOT_SUPPORT;
        }
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetServerNum(const std::vector<RankInfo_t> &ranks)
{
    std::vector<std::string> serverIds;
    for (u32 index = 0; index < ranks.size(); index++) {
        std::vector<std::string>::iterator found = find(serverIds.begin(), serverIds.end(), ranks[index].serverId);
        if (found == serverIds.end()) {
            serverIds.push_back(ranks[index].serverId);
        }
    }
    serverNum_ = serverIds.size();
    return HCCL_SUCCESS;
}

u32 HcclImplBase::GetSuperPodNum(const std::vector<RankInfo> &ranks) const
{
    u32 superPodNum = 0;
    CHK_PRT_RET(deviceType_ != DevType::DEV_TYPE_910_93, HCCL_DEBUG("deviceType is not match superPod."), superPodNum);

    std::set<std::string> superPodIds;
    for (u32 index = 0; index < ranks.size(); index++) {
        // superPodId为空时, 返回超节点数量为0, 按照非超节点模式处理
        CHK_PRT_RET(ranks[index].superPodId.empty(),
            HCCL_DEBUG("ranks[%u] superPodId[%s] is empty, set superPodNum to zero",
            index, ranks[index].superPodId.c_str()), superPodNum);

        if (superPodIds.find(ranks[index].superPodId) == superPodIds.end()) {
            superPodIds.insert(ranks[index].superPodId);
        }
    }
    superPodNum = superPodIds.size();
    return superPodNum;
}

HcclResult HcclImplBase::GetServerId(const RankTable_t &rankTable)
{
    for (u32 i = 0; i < rankTable.rankList.size(); i++) {
        if (rankTable.rankList[i].rankId == userRank_) {
            serverId_ = rankTable.rankList[i].serverId;
            superPodId_ = rankTable.rankList[i].superPodId;
            if (rankTable.rankList[i].deviceInfo.devicePhyId == HOST_DEVICE_ID) {
                isCpuRank_ = true;
            }
            break;
        }
    }
    if (serverId_.empty()) {
        HCCL_ERROR("[Get][ServerId]GetServerId fail");
        return HCCL_E_PARA;
    }
    return HCCL_SUCCESS;
}

// world group适配获取server内设配数
HcclResult HcclImplBase::GetInnerServerAverageDevice(const RankTable_t &rankTable)
{
    deviceNumPerServer_ = 0;
    bool isConnectedWithHCCS = true;
    LinkTypeInServer linkType = LinkTypeInServer::HCCS_TYPE;
    for (u32 i = 0; i < rankTable.rankList.size(); i++) {
        // 同一server的标识IP 是一样的，所以可以以此推算出平均dev个数
        if (rankTable.rankList[i].deviceInfo.devicePhyId == HOST_DEVICE_ID && isHaveCpuRank_ != true) {
            isHaveCpuRank_ = true;
        }
        if (serverId_ == rankTable.rankList[i].serverId &&
            rankTable.rankList[i].deviceInfo.devicePhyId != HOST_DEVICE_ID) {
            deviceNumPerServer_++;
        } else {
            continue;
        }
        if (isHeterogComm_ || Is310PDevice()) {
            // 异构场景无需获取链路类型并校验
            continue;
        }
        for (u32 j = i + 1; j < rankTable.rankList.size(); j++) {
            if (rankTable.rankList[i].serverId == rankTable.rankList[j].serverId) {
                if (rankTable.rankList[i].deviceInfo.devicePhyId == HOST_DEVICE_ID ||
                    rankTable.rankList[j].deviceInfo.devicePhyId == HOST_DEVICE_ID) {
                    continue;
                }
                CHK_RET(hrtGetPairDeviceLinkType(rankTable.rankList[i].deviceInfo.devicePhyId,
                    rankTable.rankList[j].deviceInfo.devicePhyId, linkType));
            }
            if (linkType != LinkTypeInServer::HCCS_TYPE) {
                isConnectedWithHCCS = false;
            }
        }
    }
    if (deviceType_ == DevType::DEV_TYPE_910B && !isConnectedWithHCCS) {
        deviceNumPerAggregation_ = deviceNumPerServer_ / FACTOR_NUM_TWO;
    } else {
        deviceNumPerAggregation_ = deviceNumPerServer_;
    }
    return HCCL_SUCCESS;
}

// sub group适配获取server内设配数
HcclResult HcclImplBase::GetInnerServerAverageDevice(const std::vector<RankInfo> &rankList)
{
    deviceNumPerServer_ = 0;
    bool isConnectedWithHCCS = true;
    LinkTypeInServer linkType = LinkTypeInServer::HCCS_TYPE;
    for (u32 i = 0; i < rankList.size(); i++) {
        // 同一server的标识IP 是一样的，所以可以以此推算出平均dev个数
        if (serverId_ == rankList[i].serverId && rankList[i].devicePhyId != HOST_DEVICE_ID) {
            deviceNumPerServer_++;
        } else {
            continue;
        }
        if (isHeterogComm_ || Is310PDevice() || isHaveCpuRank_) {
            // 异构场景无需获取链路类型并校验
            continue;
        }
        for (u32 j = i + 1; j < rankList.size(); j++) {
            if (rankList[i].serverId == rankList[j].serverId) {
                CHK_RET(hrtGetPairDeviceLinkType(rankList[i].devicePhyId, rankList[j].devicePhyId, linkType));
            }
            if (linkType != LinkTypeInServer::HCCS_TYPE) {
                isConnectedWithHCCS = false;
            }
        }
    }
    if (deviceNumPerServer_ == 0 && isHaveCpuRank_) {
        deviceNumPerServer_ = 1;
    }
    if (deviceType_ == DevType::DEV_TYPE_910B && !isConnectedWithHCCS) {
        deviceNumPerAggregation_ = deviceNumPerServer_ / FACTOR_NUM_TWO;
    } else {
        deviceNumPerAggregation_ = deviceNumPerServer_;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::TransformRankInfoByServerId(
    const std::vector<RankInfo_t> &rankList, ServRankInfo_t &servRankInfo) const
{
    // 按server重新组织rank信息，便于后续校验及信息填写
    for (size_t index = 0; index < rankList.size(); ++index) {
        const RankInfo_t &rankInfo = rankList[index];
        std::string serverId = SalTrim(rankInfo.serverId);
        // 以serverID为索引，将server下的ranks放入vector
        ServRankInfo_t::iterator itr = servRankInfo.find(serverId);
        if (itr != servRankInfo.end()) {
            itr->second.push_back(rankInfo);
        } else {
            std::vector<RankInfo_t> rankInfoList;
            rankInfoList.push_back(rankInfo);
            std::pair<std::string, std::vector<RankInfo_t>> rankInfoPair(serverId, rankInfoList);
            servRankInfo.insert(rankInfoPair);
        }
    }
    // 每个server下的rank列表按设备Id从小到大的顺序排序
    for (auto &iter : servRankInfo) {
        std::sort(iter.second.begin(), iter.second.end(), CompareWithDevicePhyId);
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckNicDeploy(NICDeployment nicDeploy, DevType deviceType) const
{
    (void)deviceType;
    if (nicDeploy >= NICDeployment::NIC_DEPLOYMENT_RESERVED) {
        HCCL_ERROR("[Check][NicDeploy]errNo[0x%016llx] nicDeploy[%u] out of range[%d, %d]",
            HCCL_ERROR_CODE(HCCL_E_PARA), nicDeploy,
            static_cast<int32_t>(NICDeployment::NIC_DEPLOYMENT_HOST),
            static_cast<int32_t>(NICDeployment::NIC_DEPLOYMENT_DEVICE));
        return HCCL_E_PARA;
    }

    return HCCL_SUCCESS;
}

bool HcclImplBase::Check2N(u32 num) const
{
    if (num < 1) {
        return false;
    } else {
        return ((num & (num - 1)) == 0);
    }
}

HcclResult HcclImplBase::CheckDevCount(const u32 devNum)
{
    if (devNum > HCCL_AISERVER_DEVICE_NUM) {
        HCCL_ERROR("[Check][DevCount]errNo[0x%016llx] devNum[%u] out of range[%u, %u]", HCCL_ERROR_CODE(HCCL_E_PARA),
            devNum, 0, HCCL_AISERVER_DEVICE_NUM);
        return HCCL_E_PARA;
    }
    // 其他拓扑算法设备数目: 1 server: 1, 2, 4, 8
    //                     n server: 1*n, 2*n, 4*n, 8*n
    if (!Check2N(devNum)) {
        const std::string devnumError("devNum must be divisible by 8, or equal to 1, 2 or 4, please check devNum");
        RPT_ENV_ERR(true,
            "EI0004",
            std::vector<std::string>({"error_reason", "ranktable_path"}),
            std::vector<std::string>({
                devnumError,
                "The ranktable path configured in the training can be "
                "found in the plogs."
                }));

        HCCL_ERROR("[Check][DevCount]errNo[0x%016llx] devNum[%u] devNum must be divisible by 8, or equal to 1, 2 or 4",
            HCCL_ERROR_CODE(HCCL_E_PARA),
            devNum);
        return HCCL_E_PARA;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckRankTable(const RankTable_t &rankTable, const ServRankInfo_t &servRankInfo)
{
    // 检查网卡挂载位置
    if (CheckNicDeploy(rankTable.nicDeploy, deviceType_) != HCCL_SUCCESS) {
        HCCL_ERROR("[Check][RankTable]errNo[0x%016llx] nicDeploy[%d] out of range[%d, %d]",
            HCCL_ERROR_CODE(HCCL_E_PARA), rankTable.nicDeploy,
            static_cast<int32_t>(NICDeployment::NIC_DEPLOYMENT_HOST),
            static_cast<int32_t>(NICDeployment::NIC_DEPLOYMENT_DEVICE));
        return HCCL_E_PARA;
    }

    if (isHeterogComm_ || Is310PDevice() || commWorkMode_ ==  WorkMode::HCCL_MODE_AI_CPU) {
        // 异构场景无需检查server内device个数
        return HCCL_SUCCESS;
    }

    // 检查服务器上的设备信息
    ServRankInfo_t::const_iterator iterBegin = servRankInfo.begin();
    u32 devNum = 0;
    CHK_RET(GetDevNum(iterBegin->second, devNum));

    bool multiServerDiffDeviceNumMode = false;
    for (ServRankInfo_t::const_iterator iter = iterBegin; iter != servRankInfo.end(); ++iter) {
        // 检测每个服务器内的设备数是否相等，如果不相同即为多server不同卡模式
        u32 curServerDevNum = 0;
        CHK_RET(GetDevNum(iter->second, curServerDevNum));
        if (devNum != curServerDevNum) {
            HCCL_WARNING("[Check][RankTable] devnum isn't same,(serverA:[%s],serverB:[%s])"\
                "devNum(%u, %u)", iterBegin->first.c_str(), iter->first.c_str(), devNum, curServerDevNum);
            multiServerDiffDeviceNumMode = true;
        }
    }

    // 非多server不同卡模式下，判断实际设备数目和rank_table中的记录一致
    if (multiServerDiffDeviceNumMode == false && rankTable.deviceNum != devNum * servRankInfo.size()) {
        HCCL_WARNING("[Check][RankTable]errNo[0x%016llx] devnum  isn't same, number in rankTable:[%u], actual:[%llu]",
            HCCL_ERROR_CODE(HCCL_E_PARA), rankTable.deviceNum, devNum * servRankInfo.size());
        return HCCL_E_PARA;
    }

    // 910模组：服务器内设备的数目必须是2的次幂,在此check(非模组形态无此限制不check)
    // 910B、910_93模组形态未定，服务器内设备的数目校验规则后续补充
    if (pairLinkInfo_[static_cast<u32>(LinkTypeInServer::HCCS_TYPE)].size() > 0 && devNum > HCCL_DEVICE_NUM_TWO &&
        (deviceType_ != DevType::DEV_TYPE_910B && deviceType_ != DevType::DEV_TYPE_910_93)) {
        CHK_PRT_RET(CheckDevCount(devNum) != HCCL_SUCCESS,
            HCCL_ERROR("[Check][RankTable]errNo[0x%016llx] devnum  is invalid in server.",
                HCCL_ERROR_CODE(HCCL_E_PARA)), HCCL_E_PARA);
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckDevPhyId(const s32 &devicePhyId) const
{
    if (devicePhyId > MAX_DEVICE_ID && devicePhyId != HOST_DEVICE_ID) {
        HCCL_ERROR("[Check][DevPhyId]errNo[0x%016llx] devicePhyId[%d] out of range[-1, %d]",
            HCCL_ERROR_CODE(HCCL_E_PARA), devicePhyId, MAX_DEVICE_ID);
        return HCCL_E_PARA;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::SortRankInfoList()
{
    // 按rank id从小到大的顺序返回
    std::sort(rankInfoList_.begin(), rankInfoList_.end(), CompareWithUserRank);

    for (u32 index = 0; index < rankInfoList_.size(); ++index) {
        CHK_PRT_RET((index != rankInfoList_[index].userRank),
            HCCL_ERROR("[HcclImplBase][SortRankInfoList]errNo[0x%016llx] index[%u] != rankInfoList.userRank[%u]",
                HCCL_ERROR_CODE(HCCL_E_PARA), index, rankInfoList_[index].userRank), HCCL_E_PARA);
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetRankInfoList(const RankTable_t &rankTable)
{
    // 检查rank table入参正确性
    CHK_RET(CheckRankTable(rankTable, servRankInfo_));
    CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());

    // 获取芯片类型
    DevType devType = DevType::DEV_TYPE_COUNT;
    CHK_RET(hrtHalGetDeviceType(0, devType));
    HCCL_INFO("[GetRankInfoList]get device type is[%lld]", devType);

    // 遍历rank table获取rank信息
    rankInfoList_.clear();
    for (auto iter = servRankInfo_.begin(); iter != servRankInfo_.end(); ++iter) {
        for (u32 index = 0; index < iter->second.size(); ++index) {
            const RankInfo_t &orgRankInfo = iter->second[index];
            // 构建comm 使用的rank 信息
            RankInfo rankInfo;
            rankInfo.userRank = orgRankInfo.rankId;
            rankInfo.worldRank = orgRankInfo.rankId;

            rankInfo.deviceType = devType;
            CHK_RET(CheckDeviceType(devType));

            if (devType != DevType::DEV_TYPE_910B || deviceType_ != DevType::DEV_TYPE_910_93) {
                // 910B、910_93形态不做devicePhyId最大值的判断
                CHK_RET(CheckDevPhyId(orgRankInfo.deviceInfo.devicePhyId));
            }
            rankInfo.devicePhyId = orgRankInfo.deviceInfo.devicePhyId;

            rankInfo.serverId = orgRankInfo.serverId;
            rankInfo.serverIdx = orgRankInfo.serverIdx;
            rankInfo.hostIp = orgRankInfo.hostIp;
            rankInfo.hostPort = orgRankInfo.hostPort;
            rankInfo.localRank = orgRankInfo.localRank;
            rankInfo.superPodId = orgRankInfo.superPodId;
            CHK_RET(GetNicInfo(rankTable.nicDeploy, index, iter->second, rankInfo));
            rankInfo.nicIdx.assign(nicList_.begin(), nicList_.end());
            rankInfoList_.push_back(rankInfo);
        }
    }
    // 将rank id从小到大的顺序返回
    CHK_RET(SortRankInfoList());

    return HCCL_SUCCESS;
}

u32 HcclImplBase::CalMeshAggRankSize(int halfDevNum) const
{
    u32 size = INVALID_VALUE_RANKSIZE;
    for (auto iter = servRankInfo_.begin(); iter != servRankInfo_.end(); ++iter) {
        u32 aggregationRankSize0 = 0;
        u32 aggregationRankSize1 = 0;
        for (u32 index = 0; index < iter->second.size(); ++index) {
            const RankInfo_t &orgRankInfo = iter->second[index];
            if (orgRankInfo.deviceInfo.devicePhyId < halfDevNum) {
                aggregationRankSize0++;
            } else {
                aggregationRankSize1++;
            }
        }
        u32 tmpsize = INVALID_VALUE_RANKSIZE;
        if (aggregationRankSize0 && aggregationRankSize1) {
            tmpsize = aggregationRankSize0;
        } else {
            tmpsize = iter->second.size();
        }
        size = size > tmpsize ? tmpsize : size;
    }
    return size;
}

HcclResult HcclImplBase::CalAndSetMeshAggRankSize()
{
    u32 size = INVALID_VALUE_RANKSIZE;
    if ((deviceType_ == DevType::DEV_TYPE_910B) && isDiffDeviceModule_) { // 910B 16p场景
        size = CalMeshAggRankSize(HCCL_DEVICE_NUM_EIGHT);
    } else if (deviceType_ == DevType::DEV_TYPE_910) {
        if (pairLinkInfo_[static_cast<u32>(LinkTypeInServer::HCCS_TYPE)].size() == 0) { // 标卡
            size = 1;
        } else { // 模组
            size = CalMeshAggRankSize(HCCL_DEVICE_NUM_FOUR);
        }
    } else { // 910B的8卡、310P 直接返回server内的size数量
        size = servRankInfo_.begin()->second.size();
    }
    CHK_RET(SetMeshAggregationRankSize(size));
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetAlgType(AlgType &algType, HcclCMDType opType)
{
    if (commWorkMode_ == WorkMode::HCCL_MODE_AI_CPU) {
        algType.algoLevel0 = AlgTypeLevel0::ALG_LEVEL0_WHOLE_RING;
        algType.algoLevel1 = AlgTypeLevel1::ALG_LEVEL1_WHOLE_RING;
        return HCCL_SUCCESS;
    }
    CHK_SMART_PTR_NULL(implAlg_);
    return implAlg_->GetAlgType(algType, opType);
}

u32 HcclImplBase::GetRankTableCrc()
{
    return ranktableCrc_;
}

void HcclImplBase::CollectTopoAttr(HcclTopoAttr &topoAttr)
{
    topoAttr.serverNum = serverNum_;
    topoAttr.superPodNum = superPodNum_;
    topoAttr.moduleNum = moduleNum_;
    topoAttr.deviceNumPerServer = deviceNumPerServer_;
    topoAttr.deviceNumPerAggregation = deviceNumPerAggregation_;
    topoAttr.multiModuleDiffDeviceNumMode = multiModuleDiffDeviceNumMode_;

    topoAttr.meshAggregationRankSize = meshAggregationRankSize_;
    topoAttr.isDiffDeviceModule = isDiffDeviceModule_;
    topoAttr.isSingleMeshAggregation = isSingleMeshAggregation_;
    topoAttr.isAllRankSamePlane = isAllRankSamePlane_;

    topoAttr.userRank = userRank_;
    topoAttr.realUserRank = realUserRank_;
    topoAttr.userRankSize = userRankSize_;

    topoAttr.devicePhyId = devicePhyId_;
    topoAttr.deviceLogicId = deviceLogicId_;
    topoAttr.deviceType = deviceType_;
    topoAttr.isStandardCard = isStandardCard_;

    topoAttr.nicList = nicList_;
    topoAttr.pairLinkCounter = pairLinkCounter_;
    topoAttr.pairLinkInfo = pairLinkInfo_;
    topoAttr.rankInfoList = rankInfoList_;
    topoAttr.isSupportRdmaLite = isSupportRdmaLite_;
}

void HcclImplBase::CollectAlgoAttr(HcclAlgoAttr &algoAttr)
{
    algoAttr.isHaveCpuRank = isHaveCpuRank_;
    algoAttr.inlineReduceSwitchOn = inlineReduceSwitchOn_;
    algoAttr.isUsedRdmaLevel0 = isUsedRdmaLevel0_;
    algoAttr.isUsedInterHccsMode = isUsedInterHccsMode_;

    algoAttr.identifier = identifier_;
    algoAttr.collectiveId = collectiveId_;

    algoAttr.nicDeployment = nicDeployment_;
    algoAttr.commWorkMode = commWorkMode_;
}

// a+x mesh间需要同时保证ip有效和roce开关打开才能走rdma
bool HcclImplBase::IsUsedRdmaLevel0AndIpInvalid()
{
    u32 nicNum = devIpAddr_.size();
    bool ipInvalid = true;
    for (u32 i = 0; i < nicNum; i++) {
        if (devIpAddr_[i].IsInvalid()) {
            HCCL_INFO("[Init][Nic]nic num[%u] deviceip is invalid, total nicNum[%u]", i, nicNum);
            ipInvalid = false;
            continue;
        }
    }
    // 机间卡数不一致场景下，IP有效情况下就走RDMA
    // 机间卡数一致场景下，需环境变量ROCE打开(IsEnableRoce在针对多机下未对此开关进行拦截)且IP有效情况下走RDMA
    return ((GetExternalInputIntraRoceSwitch() || multiModuleDiffDeviceNumMode_) && ipInvalid);
}

HcclResult HcclImplBase::InitPara(const std::string &colectiveId)
{
    // 检查当前user_rank 对应的devid和rt查到的一致
    for (u32 i = 0; i < rankInfoList_.size(); ++i) {
        if ((userRank_ == rankInfoList_[i].userRank) &&
            (static_cast<s32>(devicePhyId_) != rankInfoList_[i].devicePhyId)) {
            HCCL_ERROR("[Init][Para]errNo[0x%016llx] parameter check failed,userrank[%u] == rankInfoList.userrank[%u],"\
                "phyid[%d] != rankInfoList.devid[%d]", HCCL_ERROR_CODE(HCCL_E_PARA), userRank_,
                rankInfoList_[i].userRank, static_cast<s32>(devicePhyId_), rankInfoList_[i].devicePhyId);
            return HCCL_E_PARA;
        }
    }

    // 获取超节点数量, 大于0时识别为超节点模式
    superPodNum_ = GetSuperPodNum(rankInfoList_);
    HCCL_INFO("InitPara: serverNum[%u], superPodNum[%u]", serverNum_, superPodNum_);
    // 超节点HCCS模式
    if (deviceType_ == DevType::DEV_TYPE_910_93 && GetExternalInputInterHccsDisable() == false &&
        serverNum_ > 1 && superPodNum_ > 0) {
        isUsedInterHccsMode_ = true;
        HCCL_RUN_INFO("InitPara: will use inter HCCS Mode, superPodId[%s], superPodNum[%u], "\
            "serverNum[%u], userRank[%u]", superPodId_.c_str(), superPodNum_, serverNum_, userRank_);
    }

    collectiveId_ = colectiveId;

    if (!isHeterogComm_ && commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU) {
        CHK_RET(CalAndSetMeshAggRankSize());
    }

    // emedding service aicpu侧流水执行也需要dispatcher_去申请notify
    if (!isHeterogComm_) {
        CHK_RET(InitProfiler());
        // 根据设备ID创建dispatcher
        if ((deviceType_ == DevType::DEV_TYPE_910B || deviceType_ == DevType::DEV_TYPE_910_93) &&
            GetExternalInputHcclEnableFfts()) {
            CHK_PRT_CONT(GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE,
                HCCL_RUN_INFO("Will use ffts mode."));
        } else {
            // 不满足ffts+特性开启条件。
            SetFftsSwitch(false);
        }

        CHK_RET(HcclDispatcherInit(DispatcherType::DISPATCHER_NORMAL, devicePhyId_, &dispatcher_));
        CHK_SMART_PTR_NULL(dispatcher_);

        CHK_RET(HcclDispatcherInit(DispatcherType::DISPATCHER_VIRTURAL, devicePhyId_, &vDispatcher_));
        CHK_SMART_PTR_NULL(vDispatcher_);
    }

    notifyPool_.reset(new (std::nothrow) NotifyPool());
    CHK_SMART_PTR_NULL(notifyPool_);
    CHK_RET(notifyPool_->Init(devicePhyId_));

    queueNotifyManager_.reset(new (std::nothrow) QueueNotifyManager());
    CHK_SMART_PTR_NULL(queueNotifyManager_);
    CHK_RET(queueNotifyManager_->Init());

    callbackTask_.reset(new (std::nothrow) HcclCallbackTask(devicePhyId_, deviceLogicId_,
        dispatcher_, nicDeployment_));
    CHK_SMART_PTR_NULL(callbackTask_);

    workSpaceRes_.reset(new (std::nothrow) WorkspaceResource(devicePhyId_, deviceLogicId_, &cclBufferManager_));
    CHK_SMART_PTR_NULL(workSpaceRes_);

    if (commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU) {
        HcclTopoAttr topoAttr{};
        CollectTopoAttr(topoAttr);

        HcclAlgoAttr algoAttr{};
        CollectAlgoAttr(algoAttr);

        implAlg_ = new (std::nothrow) HcclAlg(cclBufferManager_, dispatcher_, vDispatcher_);
        CHK_SMART_PTR_NULL(implAlg_);
        CHK_RET(implAlg_->Init(static_cast<const void*>(&transportResourceInfo_), sizeof(transportResourceInfo_),
            workSpaceRes_, notifyPool_, netDevCtxMap_,
            queueNotifyManager_, algoAttr, topoAttr, isHeterogComm_));
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitHeterogRecvExecutor() const
{
    std::vector<socket_wlist_info_t> whiteList(userRankSize_);
    for (u32 i = 0; i < rankInfoList_.size(); i++) {
        whiteList[i].remote_ip.addr = rankInfoList_[i].nicIp[0].GetBinaryAddress().addr;
        whiteList[i].remote_ip.addr6 = rankInfoList_[i].nicIp[0].GetBinaryAddress().addr6;
        whiteList[i].conn_limit = CONN_LIMIT;
    }

    RaResourceInfo raResourceInfo;
    CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
    SocketHandle nicSocketHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicSocketHandle;

    HCCL_DEBUG("ip[%s] device[%d]", devIpAddr_[0].GetReadableAddress(), deviceLogicId_);

    CHK_RET(hrtRaSocketWhiteListAdd(nicSocketHandle, whiteList.data(), userRankSize_));

    return HCCL_SUCCESS;
}

// 集群中存在910B A+X时，0-7卡: moduleIdx = 2 * serverIdx; 8-15卡: moduleIdx = 2 * serverIdx + 1
// 集群中不存在910B A+X时，moduleIdx = serverIdx
HcclResult HcclImplBase::GetModuleIdx(const RankInfo_t &rankInfo, u32 &moduleIdx)
{
    CHK_PRT_RET(rankInfo.serverIdx == INVALID_UINT,
        HCCL_ERROR("serverIdx is invalid:[%u], rankId:[%u]", rankInfo.serverIdx, rankInfo.rankId), HCCL_E_PARA);
    CHK_PRT_RET(deviceType_ == DevType::DEV_TYPE_COUNT,
        HCCL_ERROR("deviceType_ is invalid:[%d], rankId:[%u]", deviceType_, rankInfo.rankId), HCCL_E_PARA);
    u32 serverIdx = rankInfo.serverIdx;
    if (deviceType_ == DevType::DEV_TYPE_910B && isDiffDeviceModule_) {
        moduleIdx = serverIdx * FACTOR_NUM_TWO + rankInfo.deviceInfo.devicePhyId / DEVICE_PER_MODULE;
    } else {
        moduleIdx = serverIdx;
    }
    CHK_PRT_RET(moduleIdx == INVALID_UINT,
        HCCL_ERROR("GetModuleIdx failed. moduleIdx:[%d], rankId:[%u]", moduleIdx, rankInfo.rankId), HCCL_E_PARA);
    return HCCL_SUCCESS;
}

// module表示通信域Inner中的节点, moduleNum即Inner中的节点个数。
// 910A:module等效于server  910BA+K: module等效于server  910B A+X:一个server包含两个module
HcclResult HcclImplBase::GetModuleInfo(const std::vector<RankInfo_t> &rankList)
{
    isDiffDeviceModule_ = IsDiffDeviceModule(rankList);
    multiModuleDiffDeviceNumMode_ = false;
    moduleNum_ = serverNum_;

    std::map<u32, std::vector<RankInfo_t>> moduleMap;
    for (RankInfo_t rankInfo : rankList) {
        u32 moduleIdx = INVALID_UINT;
        CHK_RET(GetModuleIdx(rankInfo, moduleIdx));
        if (static_cast<s32>(rankInfo.deviceInfo.devicePhyId) == HOST_DEVICE_ID) {
            continue;
        }
        auto iter = moduleMap.find(moduleIdx);
        if (iter == moduleMap.end()) {
            std::vector<RankInfo_t> rankInfoList;
            rankInfoList.push_back(rankInfo);
            moduleMap.insert(std::make_pair(moduleIdx, rankInfoList));
        } else {
            iter->second.push_back(rankInfo);
        }
    }

    // 无NPU参与通信
    if (moduleMap.size() == 0) {
        return HCCL_SUCCESS;
    }
    moduleNum_ = moduleMap.size();
    u32 preDeviceNum = moduleMap.begin()->second.size();
    u32 curDeviceNum = preDeviceNum;
    for (auto &moduleInfo: moduleMap) {
        curDeviceNum = moduleInfo.second.size();
        if (curDeviceNum != preDeviceNum) {
            multiModuleDiffDeviceNumMode_ = true;
            HCCL_RUN_INFO("different module contains different numbers of cards:[%d]", multiModuleDiffDeviceNumMode_);
        }
        HCCL_INFO("module[%d] contains [%d]devices", moduleInfo.first, moduleInfo.second.size());
        for (auto &rankInfo : moduleInfo.second) {
            HCCL_INFO("moduleIdx[%d] Info: rankId[%d], serverId[%s], serverIdx[%d], devicePhyId[%d]",
                      moduleInfo.first, rankInfo.rankId, rankInfo.serverId.c_str(), rankInfo.serverIdx,
                      rankInfo.deviceInfo.devicePhyId);
        }
    }
    return HCCL_SUCCESS;
}

// 910B A+X 在RDMA未启用情况下，两模块间的device数目需要一致且两模块中使用的卡都在同一平面上
HcclResult HcclImplBase::CheckSingleServerComm(const std::vector<RankInfo_t> &rankList) const
{
    if (serverNum_ == 1 && moduleNum_ == HCCL_MODULE_NUM_TWO && GetExternalInputIntraRoceSwitch() == 0) {
        std::vector<u32> devIdList0;
        std::vector<u32> devIdList1;
        for (RankInfo_t rankInfo : rankList) {
            if (rankInfo.deviceInfo.devicePhyId == HOST_DEVICE_ID) {
                continue;
            }
            if (rankInfo.deviceInfo.devicePhyId < DEVICE_PER_MODULE) {
                devIdList0.push_back(rankInfo.deviceInfo.devicePhyId);
            } else {
                devIdList1.push_back(rankInfo.deviceInfo.devicePhyId);
            }
        }
        std::sort(devIdList0.begin(), devIdList0.end());
        std::sort(devIdList1.begin(), devIdList1.end());

        if (devIdList0.size() != devIdList1.size()) {
            HCCL_ERROR("[Check][SingleServerComm]errNo[0x%016llx]. In A+X serverNum_[%d], moduleNum_[%d] case: "\
                "deviceNum in module0:[%d] not equal to deviceNum in module1:[%d]",
                HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT),  serverNum_, moduleNum_, devIdList0.size(), devIdList1.size());
            return HCCL_E_NOT_SUPPORT;
        }
        for (size_t i = 0; i < devIdList0.size(); i++) {
            if (devIdList0[i] % DEVICE_PER_MODULE != devIdList1[i] % DEVICE_PER_MODULE) {
                HCCL_ERROR("[Check][SingleServerComm]errNo[0x%016llx]. In A+X serverNum_[%d], moduleNum_[%d] case: "\
                    "deviceId[%d] in module0 and deviceId[%d] in module1 are not on the same plane",
                    HCCL_ERROR_CODE(HCCL_E_NOT_SUPPORT), serverNum_, moduleNum_, devIdList0[i], devIdList1[i]);
                return HCCL_E_NOT_SUPPORT;
            }
        }
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitHdcWhlistAddImpl(const s32 deviceLogicId, const HcclIpAddress &ipAddr,
    std::vector<socket_wlist_info_t>& whiteList)
{
    SocketHandle nicSocketHandle;
    RaResourceInfo raResourceInfo;

    CHK_RET(NetworkManager::GetInstance(deviceLogicId).GetRaResourceInfo(raResourceInfo));
    CHK_PRT_RET(raResourceInfo.nicSocketMap.find(ipAddr) == raResourceInfo.nicSocketMap.end(),
        HCCL_ERROR("Get socket handle for ip[%s] failed", ipAddr.GetReadableAddress()),
        HCCL_E_PARA);
    nicSocketHandle = raResourceInfo.nicSocketMap[ipAddr].nicSocketHandle;

    CHK_RET(hrtRaSocketWhiteListAdd(nicSocketHandle, whiteList.data(), whiteList.size()));
    HCCL_INFO("InitHdcWhlistAdd nicSocketHandle for ip[%s], deviceLogicId[%d] success",
        ipAddr.GetReadableAddress(), deviceLogicId);
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitHdcWhlistAdd()
{
    std::vector<RankInfo> *rankListPtr = worldRankInfoList_.size() > 0 ? &worldRankInfoList_ : &rankInfoList_;
    std::vector<socket_wlist_info_t> whiteList(rankListPtr->size());
    constexpr u32 connLimit = CONN_LIMIT;

    for (u32 i = 0; i < rankListPtr->size(); i++) {
        whiteList[i].remote_ip.addr = (*rankListPtr)[i].nicIp[0].GetBinaryAddress().addr;
        whiteList[i].remote_ip.addr6 = (*rankListPtr)[i].nicIp[0].GetBinaryAddress().addr6;
        whiteList[i].conn_limit = connLimit;
        HCCL_DEBUG("InitHdcWhlistAdd ip[%s]", (*rankListPtr)[i].nicIp[0].GetReadableAddress());
    }

    if (role_ == SERVER_ROLE_SOCKET || static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
        // ps侧以device网卡为粒度添加白名单
        for (const auto &rankInfo : rankInfoList_) {
            if (hostIp_ == rankInfo.hostIp && rankInfo.devicePhyId != HOST_DEVICE_ID) {
                u32 devLogicId;
                CHK_RET(hrtGetDeviceIndexByPhyId(rankInfo.devicePhyId, devLogicId));
                CHK_RET(InitHdcWhlistAddImpl(devLogicId, rankInfo.nicIp[0], whiteList));
            }
        }
        return HCCL_SUCCESS;
    }

    if (IsEnableRoce()) {
        // worker侧在HcclImplBase::Init中只有该标志位为true时才会初始化网卡，否则直接添加白名单找不到通信句柄
        CHK_RET(InitHdcWhlistAddImpl(deviceLogicId_, devIpAddr_[0], whiteList));
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::Init(HcclCommParams &params, const RankTable_t &rankTable)
{
    CHK_RET(InitCommParams(params));

    CHK_RET(InitTcpMode(rankTable));

    // 获取serverId
    CHK_RET(GetServerId(rankTable));

    // 获取server数
    CHK_RET(GetServerNum(rankTable.rankList));

    // 获取server内设备数
    CHK_RET(GetInnerServerAverageDevice(rankTable));

    // 根据server整理rank信息
    CHK_RET(TransformRankInfoByServerId(rankTable.rankList, servRankInfo_));

    // 获取module相关信息，moduleNum_, isDiffDeviceModule_, multiModuleDiffDeviceNumMode_;
    CHK_RET(GetModuleInfo(rankTable.rankList));

    // 生成nicList
    for (auto &iter : servRankInfo_[serverId_]) {
        if (((!iter.hostIp.IsInvalid()) || (!iter.deviceInfo.deviceIp[0].IsInvalid())) &&
            (iter.deviceInfo.devicePhyId != HOST_DEVICE_ID)) {
            nicList_.push_back(iter.deviceInfo.devicePhyId);
        }
    }
    std::sort(nicList_.begin(), nicList_.end());
    if (!isHeterogComm_ && commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU && !isCpuRank_) {
        topoInfoParse_.reset(new (std::nothrow) TopoInfoParse());
        CHK_SMART_PTR_NULL(topoInfoParse_);
        CHK_RET(topoInfoParse_->Init(rankTable, serverId_, deviceNumPerServer_));
        CHK_RET(topoInfoParse_->GetServerInnerLinkInfo(pairLinkCounter_, pairLinkInfo_));   // 获取本Server上HCCS、PXI链接的数目
        // 初始化阶段判断组网状态
        CHK_RET(topoInfoParse_->IsSingleMeshAggregation(isSingleMeshAggregation_));         // 确认集群中只有一个MeshAggregation
        CHK_RET(topoInfoParse_->IsAllRankSamePlane(isAllRankSamePlane_));                   // 确认集群所有卡在一个平面上
        isStandardCard_ = IsStandardCard();
        CHK_RET(topoInfoParse_->ParseAndCheck(nicList_));
    }

    // 校验A+X单机双module场景下通信能否建立
    CHK_RET(CheckSingleServerComm(rankTable.rankList));

    // 解析ranktable信息(生成rankInfoList_)，供给commfactory使用
    CHK_RET(GetRankInfoList(rankTable));

    for (u32 i = 0; i < rankInfoList_.size(); i++) {
        HCCL_DEBUG(" host ip: %s host port: %u dev phy id: %d.", rankInfoList_[i].hostIp.GetReadableAddress(),
            rankInfoList_[i].hostPort, rankInfoList_[i].devicePhyId);
        if (rankInfoList_[i].userRank == userRank_) {
            devIpAddr_ = rankInfoList_[i].nicIp;
            devicePhyId_ = rankInfoList_[i].devicePhyId;
            hostIp_ = rankInfoList_[i].hostIp;
            hostPort_ = rankInfoList_[i].hostPort;
            localRank_ = rankInfoList_[i].localRank;
            HCCL_DEBUG("localRank_[%u].", localRank_);
            break;
        }
    }

    ranksPort_.resize(userRankSize_, 0);
    for (auto &rankInfo : rankTable.rankList) {
        ranksPort_[rankInfo.rankId] = rankInfo.deviceInfo.port == HCCL_INVALID_PORT || rankInfo.deviceInfo.port == 0 ?
            HETEROG_CCL_PORT : rankInfo.deviceInfo.port;
    }

    for (auto &rankInfo : rankInfoList_) {
        if (rankInfo.devicePhyId == HOST_DEVICE_ID) {
            CHK_RET(GetDeviceNicInfo(rankInfo));
        }
    }
    // 在确定 servRankInfo_ 和 serverId_ 信息后，就完成初始判断

    HCCL_DEBUG("HcclImplBase::Init deviceType[%u].", deviceType_);
    if (isHeterogComm_) {
        CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
        HCCL_INFO("init heterog comm, rank id[%u] device id[%u]", userRank_, devicePhyId_);

        HcclResult ret = InitPara(rankTable.collectiveId);

        CHK_PRT_RET(ret != HCCL_SUCCESS,
            HCCL_ERROR("[HcclImplBase][Init]errNo[0x%016llx] collectiveid[%s] parameter initialization failed",
            HCCL_ERROR_CODE(ret), rankTable.collectiveId.c_str()), ret);

        CHK_PRT_RET(devIpAddr_.empty(), HCCL_ERROR("[HcclImplBase][Init]devIpAddr_ size[%llu] "
            "should be greater than 0.", devIpAddr_.size()), HCCL_E_UNAVAIL);

        if (params.attr.mode != WorkMode::HCCL_MODE_AI_CPU && params.attr.mode != WorkMode::HCCL_MODE_PS) {
            HCCL_DEBUG("attr mode is not HCCL_MODE_AI_CPU nor HCCL_MODE_PS");
            CHK_RET(InitHeterogRaResource(rankTable));
            CHK_RET(MrManagerInit());
            CHK_RET(InitRecvMsgAndRequestBuffer());
            CHK_RET(InitMemBlocksAndRecvWrMem());
            CHK_RET(CreateSrq());
        }

        if (GetExternalInputHcclIsTcpMode()) {
            TcpSendThreadPool::GetSendPoolInstance()->Init(devicePhyId_);
        }
    } else {
        if (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
            deviceLogicId_ = 0;
        } else {
            CHK_RET(hrtGetDevice(&deviceLogicId_));
        }

        CHK_RET(InitPreResource(rankTable));

        interServer_ = rankTable.serverNum > 1; // serverNum为1时，不进行roce初始化
        nicDeployment_ = rankTable.nicDeploy;
        CHK_RET(InitRaResource());

        CHK_RET(InitProfiling());

        std::unique_lock<std::mutex> lock(g_hcomInitMutex);
        CHK_RET(RegistTaskExceptionHandler());

        std::string collectiveId = rankTable.collectiveId.empty() ? params.id.internal : rankTable.collectiveId;

        // 初始化参数(需要放置在ranktable解析之后)
        HcclResult ret = InitPara(collectiveId);
        CHK_PRT_RET(ret != HCCL_SUCCESS,
            HCCL_ERROR("[HcclImplBase][Init]errNo[0x%016llx] collectiveid[%s] parameter initialization failed",
            HCCL_ERROR_CODE(ret), params.id.internal), ret);
        lock.unlock();

        if (static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID) {
            HCCL_DEBUG("HcclImplBase::Init devicePhyId[%u] != HOST_DEVICE_ID.", devicePhyId_);
            CHK_RET(StreamActiveManager::GetInstance(deviceLogicId_).Init());
            CHK_RET(MrManagerInit());
            // server数量不为1且非TCP模式时初始化RDMA资源
            if (rankTable.serverNum != SINGLE_SERVER_NUM && !GetExternalInputHcclIsTcpMode()) {
                CHK_RET(InitRecvMsgAndRequestBuffer());
                CHK_RET(InitMemBlocksAndRecvWrMem());
            }
        }
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::TransformRankList(
    const std::vector<RankInfo> &rankListIn, std::vector<RankInfo_t> &rankListOut)
{
    for (size_t index = 0; index < rankListIn.size(); ++index) {
        RankInfo_t rankInfoTmp;
        rankInfoTmp.serverId = rankListIn[index].serverId;
        rankInfoTmp.deviceInfo.devicePhyId = rankListIn[index].devicePhyId;
        rankInfoTmp.serverIdx = rankListIn[index].serverIdx;
        rankInfoTmp.rankId = rankListIn[index].userRank;
        rankInfoTmp.hostIp = rankListIn[index].hostIp;
        rankInfoTmp.hostPort = rankListIn[index].hostPort;
        rankInfoTmp.localRank = rankListIn[index].localRank;
        rankInfoTmp.superPodId = rankListIn[index].superPodId;
        rankListOut.push_back(rankInfoTmp);
    }
    return HCCL_SUCCESS;
}

bool HcclImplBase::IsStandardCard()
{
    return ((pairLinkInfo_[static_cast<u32>(LinkTypeInServer::HCCS_TYPE)].size() == 0) &&
           (pairLinkInfo_[static_cast<u32>(LinkTypeInServer::HCCS_SW_TYPE)].size() == 0));
}

HcclResult HcclImplBase::Init(HcclCommParams &params, const std::vector<RankInfo> &rankList,
    WorldGroupInfo &groupCommonData)
{
    CHK_RET(InitCommParams(params));

    // 记录serverId
    serverId_ = groupCommonData.serverId;

    CHK_RET(SetWorldGroupInfo(groupCommonData.phyIdNicInfoMap, groupCommonData.worldRankInfoList,
        groupCommonData.ranksPort));
    for (auto &rankInfo : rankList) {
        if (rankInfo.devicePhyId == HOST_DEVICE_ID) {
            isHaveCpuRank_ = true;
            break;
        }
    }

    for (auto &rankInfo : worldRankInfoList_) {
        if (rankInfo.devicePhyId == HOST_DEVICE_ID) {
            isUseRankPort_ = true;
            break;
        }
    }

    CHK_RET(IsHostUseDevNic(isHostUseDevNic_));

    // 获取server内平均device数
    CHK_RET(GetInnerServerAverageDevice(rankList));

    // 将子通信域的ranklist结构体形式转换成全局通信域的
    std::vector<RankInfo_t> rankListNew;
    CHK_RET(TransformRankList(rankList, rankListNew));

    // 获取server数
    CHK_RET(GetServerNum(rankListNew));

    // 获取module相关信息，moduleNum_, isDiffDeviceModule_, multiModuleDiffDeviceNumMode_;
    CHK_RET(GetModuleInfo(rankListNew));

    // 根据server整理rank信息
    CHK_RET(TransformRankInfoByServerId(rankListNew, servRankInfo_));

    // 设置rank关联信息
    rankInfoList_.assign(rankList.begin(), rankList.end());
    for (u32 i = 0; i < rankInfoList_.size(); i++) {
        if (rankInfoList_[i].userRank == userRank_) {
            devIpAddr_ = rankInfoList_[i].nicIp;
            devicePhyId_ = rankInfoList_[i].devicePhyId;
            isCpuRank_ = rankInfoList_[i].devicePhyId == HOST_DEVICE_ID ? true : false;
            hostIp_ = rankInfoList_[i].hostIp;
            hostPort_ = rankInfoList_[i].hostPort;
            nicList_.assign(rankInfoList_[i].nicIdx.begin(), rankInfoList_[i].nicIdx.end());
            nicDeployment_ = rankInfoList_[i].nicDeploy;
            break;
        }
    }

    if (!isHeterogComm_ && commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU && !isCpuRank_) {
        topoInfoParse_.reset(new (std::nothrow) TopoInfoParse());
        CHK_SMART_PTR_NULL(topoInfoParse_);
        CHK_RET(topoInfoParse_->Init(rankList, serverId_, deviceNumPerServer_));
        CHK_RET(topoInfoParse_->GetServerInnerLinkInfo(pairLinkCounter_, pairLinkInfo_));
        // 初始化阶段判断组网状态
        CHK_RET(topoInfoParse_->IsSingleMeshAggregation(isSingleMeshAggregation_));
        CHK_RET(topoInfoParse_->IsAllRankSamePlane(isAllRankSamePlane_));
        isStandardCard_ = IsStandardCard();
        if (!isStandardCard_) {
            CHK_RET(topoInfoParse_->Check());
        }
    }

    //  inline reduce 开关
    inlineReduceSwitchOn_ = groupCommonData.inlineReduceSwitchOn;

    groupRanksPort_.resize(rankInfoList_.size(), 0);
    if (ranksPort_.size()) {
        for (auto &rankInfo : rankInfoList_) {
            groupRanksPort_[rankInfo.userRank] = ranksPort_[rankInfo.worldRank];
            HCCL_INFO("hostIp[%s], nicIp[%s], rankInfo.userRank[%u], rankInfo.worldRank[%u], port[%u], devicePhyId[%d]",
                rankInfo.hostIp.GetReadableAddress(), rankInfo.nicIp[0].GetReadableAddress(),
                rankInfo.userRank, rankInfo.worldRank, groupRanksPort_[rankInfo.userRank], rankInfo.devicePhyId);
        }
    }

    for (auto &rank : rankInfoList_) {
        if (hostIp_ != rank.hostIp) {
            isServerInter_ = true;
            HCCL_DEBUG(" isServerInter_ is true");
            break;
        }
    }

    if (HcclCheckLogLevel(HCCL_LOG_DEBUG)) {
        // 打印原来的nicList_
        std::ostringstream stringRepresentation;
        for (std::vector<uint32_t>::iterator it = nicList_.begin(); it != nicList_.end(); it++) {
            stringRepresentation << *it << " ";
        }
        std::string nicListString = stringRepresentation.str();
        const char *charNicList = nicListString.c_str();
        HCCL_DEBUG("[HcclImplBase][Init] The original nicList_: %s", charNicList);
    }
    // 更新成跟子通信域相关的nicList_
    std::vector<u32> subCommNicList;
    for (u32 i = 0; i < rankInfoList_.size(); i++) {
        if (rankInfoList_[i].serverId == serverId_ &&
            std::find(nicList_.begin(), nicList_.end(), rankInfoList_[i].devicePhyId) != nicList_.end()) {
            subCommNicList.push_back(rankInfoList_[i].devicePhyId);
        }
    }
    nicList_ = subCommNicList;
    if (HcclCheckLogLevel(HCCL_LOG_DEBUG)) {
        // 打印更新后的nicList_
        std::ostringstream stringRepresentation;
        for (std::vector<uint32_t>::iterator it = nicList_.begin(); it != nicList_.end(); it++) {
            stringRepresentation << *it << " ";
        }
        std::string nicListString = stringRepresentation.str();
        const char *charNicList = nicListString.c_str();
        HCCL_DEBUG("[HcclImplBase][Init] The subcommunication domain related nicList_: %s", charNicList);
    }

    interServer_ = serverNum_ > 1; // serverNum为1时，不进行roce初始化
    CHK_RET(InitRaResource());
    // 初始化参数(需要放置在ranktable解析之后)
    HcclResult ret = InitPara(params.id.internal);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[HcclImplBase][Init]errNo[0x%016llx] collectiveid[%s] parameter initialization failed",
        HCCL_ERROR_CODE(ret), params.id.internal), ret);

    if (IsEnableRoce() && commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU) {
        isUsedRdmaLevel0_ = IsUsedRdmaLevel0AndIpInvalid();
    }

    return HCCL_SUCCESS;
}
// HCCL 1.0 optimize

HcclResult HcclImplBase::DestroyCDomainResource(s32 tag)
{
    HCCL_INFO("[HcclImplBase]DestroyCDomainResource started. tag[%d]", tag);
    unique_lock<std::mutex> lockOpExecutorMap(tagOpExecutorMapMutex_);
    tagOpExecutorMap_.Erase(tag);
    lockOpExecutorMap.unlock();
    HCCL_INFO("[HcclImplBase]DestroyCDomainResource tagOpExecutorMap_ erase tag[%d]", tag);

    {
        lock_guard<shared_timed_mutex> lock(transportHeterogMap_.GetMtx());
        auto iter = transportHeterogMap_.FindLockFree(tag);
        if (iter == transportHeterogMap_.EndLockFree()) {
            return HCCL_SUCCESS;
        }

        for (auto &transportPtr : iter->second) {
            if (transportPtr.second != nullptr) {
                CHK_RET(transportPtr.second->Deinit());
            }
        }

        transportHeterogMap_.EraseLockFree(tag);
    }

    std::unique_lock<std::mutex> tagNodesInfoLock(tagNodesInfoLock_);
    tagNodesInfo_.erase(tag);
    tagNodesInfoLock.unlock();

    HCCL_INFO("[HcclImplBase]DestroyCDomainResource transportHeterogMap_ erase tag[%u]", tag);

    return HCCL_SUCCESS;
}

bool HcclImplBase::CompareWithDevicePhyId(const RankInfo_t &left, const RankInfo_t &right)
{
    return left.deviceInfo.devicePhyId < right.deviceInfo.devicePhyId;
}

bool HcclImplBase::CompareWithServerId(const ServerInfo_t &left, const ServerInfo_t &right)
{
    return (strcmp(left.serverId.c_str(), right.serverId.c_str()) < 0);
}

bool HcclImplBase::CompareWithNicName(const NetworkInfo_t &left, const NetworkInfo_t &right)
{
    return (strcmp(left.ethName.c_str(), right.ethName.c_str()) < 0);
}

bool HcclImplBase::CompareWithUserRank(const RankInfo &left, const RankInfo &right)
{
    return left.userRank < right.userRank;
}

std::string HcclImplBase::GetUniqueId(void)
{
    static std::atomic<u32> idCounter(0);

    std::string uniqueId("");
    uniqueId += std::to_string(SalGetPid());
    uniqueId += '-';
    uniqueId += std::to_string(idCounter.fetch_add(1));
    uniqueId += '-';
    uniqueId += std::to_string(SalGetSysTime());

    return uniqueId;
}

HcclResult HcclImplBase::GetNicInfo(const NICDeployment &nicDeploy, const u32 curRankIndex,
    const std::vector<RankInfo_t> &servRankList, RankInfo &rankInfo) const
{
    CHK_PRT_RET(servRankList.empty(), HCCL_ERROR("[Get][NicInfo]errNo[0x%016llx] server rank list is empty",
        HCCL_ERROR_CODE(HCCL_E_PARA)), HCCL_E_PARA);

    rankInfo.nicDeploy = nicDeploy;
    if (nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST) {
        // 检查网卡个数
        // 网卡挂载位置在host时，按rank index从网卡列表中获取
        const RankInfo_t &curRankInfo = servRankList[curRankIndex];
        rankInfo.nicIp.push_back(curRankInfo.hostIp);
    } else {
        CHK_PRT_RET(curRankIndex >= servRankList.size(), HCCL_ERROR("[Get][NicInfo]rankindex[%u] invalid,rank list "\
            "size is[%zu]", curRankIndex, servRankList.size()), HCCL_E_PARA);

        const RankInfo_t &curRankInfo = servRankList[curRankIndex];
        CHK_PRT_RET(curRankInfo.deviceInfo.deviceIp.size() == 0,
            HCCL_ERROR("[Get][NicInfo]rankindex[%u] invalid,deviceIp is zero", curRankIndex), HCCL_E_PARA);
        rankInfo.nicIp.push_back(curRankInfo.deviceInfo.deviceIp[0]);
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitCommParams(HcclCommParams &params)
{
    commHandle_ = params.commHandle;
    userRank_ = params.rank;
    realUserRank_ = params.userRank;
    userRankSize_ = params.totalRanks;
    deviceLogicId_ = params.logicDevId;
    profilingOption_ = params.profilingOption;
    profilingInitiated_ = params.profilingInitiated;
    deviceType_ = params.deviceType;
    isHeterogComm_ = params.isHeterogComm;
    commWorkMode_ = params.commWorkMode;
    hcomGroupNicInit_ = params.hcomGroupNicInit;
    identifier_ = params.identifier;
    ranktableCrc_ = params.ranktableCrc;

    if (!isHeterogComm_) {
        /* 申请trace资源信息 */
        std::string logInfo = "HCCL_";
        logInfo.append(to_string(SalGetTid()));
        logInfo.append("_");
        logInfo.append(to_string(deviceLogicId_));

        opBaseAtraceInfo_.reset(new (std::nothrow) HcclTraceInfo());
        CHK_PTR_NULL(opBaseAtraceInfo_);
        CHK_RET(opBaseAtraceInfo_->Init(logInfo));
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitPreResource(const RankTable_t &rankTable)
{
    if (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
        return HCCL_SUCCESS;
    }
    (void)rankTable;
    // 查询本rank所在服务器
    auto iterServ = servRankInfo_.find(serverId_);

    bool check = (iterServ == servRankInfo_.end());
    CHK_PRT_RET(check, HCCL_ERROR("[Init][PreResource]can't find serverId[%s] in server map", serverId_.c_str()),
        HCCL_E_NOT_FOUND);

    for (u32 i = 0; i < iterServ->second.size(); i++) {
        if (iterServ->second[i].deviceInfo.devicePhyId != HOST_DEVICE_ID) {
            enableP2PDevices_.push_back(iterServ->second[i].deviceInfo.devicePhyId);
        }
    }

    HcclResult ret = P2PMgmtPub::EnableP2P(enableP2PDevices_);
    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Init][PreResource]Enable P2P Failed, deviceLogicId[%d], ret[%u]",
        deviceLogicId_, ret), ret);

    drvInit_ = true;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitTcpMode(const RankTable_t &rankTable) const
{
    bool isTcpMode = false;
    HCCL_INFO("[TcpMode][%u] [1:TCP, 2:RDMA, 3:RESERVED].", GetExternalInputProtocolType());
    if (GetExternalInputProtocolType() == ProtocolType::TCP) {
        isTcpMode = true;
    } else if (GetExternalInputProtocolType() == ProtocolType::RDMA) {
    // 通信协议选择RDMA
    } else {
        isTcpMode = (rankTable.nicDeploy == NICDeployment::NIC_DEPLOYMENT_HOST &&
            !GetExternalInputHcclHostRdmaEnable());
        HCCL_INFO("[Init][TcpMode]isTcpMode[%d] nicDeploy[%d] hcclDeviceNicDisable[%d] hcclHostRdmaEnable[%d]",
            isTcpMode, rankTable.nicDeploy, GetExternalInputHcclDeviceNicDisable(),
            GetExternalInputHcclHostRdmaEnable());
    }
    SetTcpMode(isTcpMode);

    // 异构场景解析外部输入,放在SetTcpMode前防止Tcp用例走错分支，放在RecordProtocolType确保hdc模式下建链通信协议校验正确
    CHK_RET(InitExternalInputHeterog());
    return HCCL_SUCCESS;
}

bool HcclImplBase::IsSupportEnableRoce()
{
    // 910B单机两种使能roce场景：1、a+x同时使用两module  2.标卡
    bool roceSwitch = false;
    HCCL_INFO("[HcclImplBase]IsSupportEnableRoce log");
    if (deviceType_ == DevType::DEV_TYPE_910B) {
        roceSwitch = (GetExternalInputIntraRoceSwitch() && (!isSingleMeshAggregation_ || isStandardCard_)) ||
                     multiModuleDiffDeviceNumMode_;
    } else if (deviceType_ == DevType::DEV_TYPE_910_93) {
        roceSwitch = GetExternalInputEnableRdmaSdmaConcurrent();
    } else { // 其他单机场景为了防止用户误用roce开关
        roceSwitch = isStandardCard_ ? GetExternalInputIntraRoceSwitch() : false;
    }
    return roceSwitch;
}

bool HcclImplBase::IsEnableRoce()
{
    // 910B单机两种使能roce场景：1、a+x同时使用两module  2.标卡
    bool roceSwitch = IsSupportEnableRoce();
    HCCL_INFO("interServer_[%u] isSingleMeshAggregation_[%u] roceSwitch[%u].",
        interServer_, isSingleMeshAggregation_, roceSwitch);

    bool isInnerSuperPodHccsMode = false;
    // 单个910 93超节点内节点间走HCCS通信, 不需要使能NIC
    if (deviceType_ == DevType::DEV_TYPE_910_93 && superPodNum_ == 1 &&
        GetExternalInputInterHccsDisable() == false) {
        isInnerSuperPodHccsMode = true;
        HCCL_INFO("IsEnableRoce isInnerSuperPodHccsMode set %d", isInnerSuperPodHccsMode);
    }
    if ((interServer_ && !isInnerSuperPodHccsMode) || roceSwitch) {
        return true;
    }
    return false;
}

u32 HcclImplBase::GetLocalNicPort()
{
    if (isHaveCpuRank_) {
        isUseRankPort_ = true;
    }
    return GetNicPort(devicePhyId_, ranksPort_, userRank_, isUseRankPort_);
}

HcclResult HcclImplBase::InitRaResource()
{
    /* 本通信域内只有1个device时，不需要初始化ra资源 */
    if (userRankSize_ <= 1) {
        HCCL_INFO("user rank size <= 1, ra is not needed for single device");
        return HCCL_SUCCESS;
    }

    socketManager_.reset(new (std::nothrow) HcclSocketManager(nicDeployment_, deviceLogicId_, devicePhyId_, userRank_));
    CHK_PTR_NULL(socketManager_);
    CHK_RET(IsHostUseDevNic(isHostUseDevNic_));

    if (static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID) {
        CHK_RET(HcclNetInit(NICDeployment::NIC_DEPLOYMENT_DEVICE, devicePhyId_, deviceLogicId_, false));
    }

    if ((static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID && IsEnableRoce() &&
        nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST) ||
        (Is310PDevice() && nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST) ||
        (!Is310PDevice() && commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU && isHaveCpuRank_)) {
        u32 devicePhyID = (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) ? 0 : devicePhyId_;
        CHK_RET(HcclNetInit(NICDeployment::NIC_DEPLOYMENT_HOST, devicePhyID, deviceLogicId_, false));
    }

    socketManager_.reset(new (std::nothrow) HcclSocketManager(nicDeployment_, deviceLogicId_, devicePhyId_, userRank_));
    CHK_PTR_NULL(socketManager_);

    if (Is310PDevice()) {
        CHK_RET(InitNic());
    } else if (static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID) {
        localVnicListenPort_ = GetLocalNicPort();
        localVnicIp_ = HcclIpAddress(devicePhyId_);
        CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
                devicePhyId_, DeviceIdType::DEVICE_ID_TYPE_PHY_ID, devicePhyId_, localVnicIp_));
        HcclNetDevCtx vnicPortCtx;
        CHK_RET(HcclNetOpenDev(&vnicPortCtx, NicType::VNIC_TYPE, devicePhyId_, deviceLogicId_, localVnicIp_));
        netDevCtxMap_.insert(std::make_pair(localVnicIp_, vnicPortCtx));
        CHK_RET(socketManager_->ServerInit(vnicPortCtx, localVnicListenPort_));

        if (IsEnableRoce()) {
            CHK_RET(InitNic()); // isUsedRdmaLevel0_默认为false，若初始化网卡时，网卡IP有效才根据环境变量配置
        }
    }
    HCCL_INFO("isUsedRdmaLevel0_[%u] nicNum[%u] hostIP[%s], nicDeployment[%d].",
        isUsedRdmaLevel0_, devIpAddr_.size(), hostIp_.GetReadableAddress(), nicDeployment_);

    if (!Is310PDevice() && commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU) {
        CHK_RET(InitHeterogHostNic());
    }

    raResourceInit_ = true; // 全局通信域会初始化，子通信域不会初始化，但是析构均会进入此逻辑，需要标记
    return HCCL_SUCCESS;
}


HcclResult HcclImplBase::InitHeterogRaResource(const RankTable_t &rankTable)
{
    CHK_PRT_RET(rankTable.rankList.size() != userRankSize_, HCCL_ERROR("[Init][HeterogRaResourc] rank list size[%u]" \
        " is different from user rank size[%u]", rankTable.rankList.size(), userRankSize_), HCCL_E_PARA);
    ranksPort_.resize(userRankSize_, 0);
    for (auto &rankInfo : rankTable.rankList) {
        ranksPort_[rankInfo.rankId] = rankInfo.deviceInfo.port == HCCL_INVALID_PORT || rankInfo.deviceInfo.port == 0
            ? HETEROG_CCL_PORT : rankInfo.deviceInfo.port;
    }

    heterogRaInit_ = true;
    CHK_RET(NetworkManager::GetInstance(deviceLogicId_).HeterogInit(devicePhyId_, devIpAddr_[0],
        ranksPort_[userRank_]));
    if (!GetExternalInputHcclIsTcpMode()) {
        hostRdmaInitFlag_ = true;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitHeterogRaResource()
{
    if (heterogRaInit_) {
        HCCL_INFO("deinit heterog ra resource!");
        CHK_RET(NetworkManager::GetInstance(deviceLogicId_).HeterogDeinit(devicePhyId_, devIpAddr_[0],
            ranksPort_[userRank_]));
        heterogRaInit_ = false;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitPsRaResource(u32 devLogicId, s32 devicePhyId, const HcclIpAddress &ipAddr,
    u32 port, bool isHostUseDevNic, bool isBoardVersion)
{
    HCCL_INFO("Init ipAddr[%s], devicePhyId[%u], deviceLogicId_[%d], port[%u], isHostUseDevNic[%u]",
        ipAddr.GetReadableAddress(), devicePhyId, devLogicId, port, static_cast<u32>(isHostUseDevNic));
    CHK_RET(NetworkManager::GetInstance(devLogicId).PsWorkerRaInit(devicePhyId, ipAddr, port, isHostUseDevNic,
        GetRemoteIsHdc(), isBoardVersion));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitPsHdcRaResource()
{
    // hdc模式下基于device网卡进行ra资源的初始化
    std::string hostIp(hostIp_.GetReadableAddress());
    for (const auto &devsInfo : rankDevicePhyIdNicInfoMap_[hostIp]) {
        u32 devLogicId;
        CHK_RET(hrtGetDeviceIndexByPhyId(devsInfo.first, devLogicId));
        CHK_RET(InitPsRaResource(devLogicId, devsInfo.first, devsInfo.second, ranksPort_[realUserRank_], true));
        HcclNetDevCtx nicPortCtx;
        CHK_RET(HcclNetOpenDev(&nicPortCtx, NicType::DEVICE_NIC_TYPE, devsInfo.first, devLogicId, devsInfo.second));
        netDevCtxMap_.insert(std::make_pair(devsInfo.second, nicPortCtx));
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitPsRaResource()
{
    CHK_PRT_RET((hostIp_.IsInvalid()), HCCL_ERROR("[Init][Nic] host ip is invalid when NIC "
        "deployment is host. "), HCCL_E_PARA);
    HCCL_INFO("[Init][PsRaResource], hostPort[%u], localRank_[%u].", ranksPort_[userRank_], localRank_);

    bool hostIpInited = false;
    bool loopBackIpInited = false;

    for (const auto &clientInfo : clientsInfo_) {
        if (serverId_ != clientInfo.serverId && !hostIpInited) {
            if (isHostUseDevNic_) {
                // hdc模式需要额外初始化device网卡的ra资源
                CHK_RET(InitPsHdcRaResource());
                // hdc模式下hccp默认开启白名单校验，ps侧需要以device网卡为粒度添加白名单
                CHK_RET(InitHdcWhlistAdd());
            } else {
                CHK_RET(InitPsRaResource(deviceLogicId_, devicePhyId_, hostIp_,
                    ranksPort_[userRank_], false));
            }
            hostIpInited = true;
        } else if (!loopBackIpInited) {
            // 环回Ip以peer模式初始化ra资源
            CHK_RET(InitPsRaResource(deviceLogicId_, devicePhyId_, loopBackIp_,
                ranksPort_[userRank_], false, true));
            loopBackIpInited = true;
        }
    }

    // 跨server开启device网卡
    psRaInit_ = true;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitWorkerRaResource()
{
    s32 port = 0;
    if (ranksPort_[userRank_] != 0) {
        port = ranksPort_[userRank_];
    } else {
        if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT) {
            port = HOST_PARA_BASE_PORT + devicePhyId_;
        } else {
            port = devicePhyId_ + GetExternalInputHcclIfBasePort() + HCCL_AISERVER_DEVICE_NUM;
        }
    }
    CHK_PRT_RET((hostIp_.IsInvalid()), HCCL_ERROR("[Init][Nic] host ip is invalid when NIC "
        "deployment is host. "), HCCL_E_PARA);
    HCCL_INFO("[Init][WorkerRaResource], hostPort[%d], localRank_[%u], devicePhyId_[%u].",
        port, localRank_, devicePhyId_);
    HCCL_INFO("Init loopBackIp_[%s], deviceLogicId_[%d]!", loopBackIp_.GetReadableAddress(), deviceLogicId_);
    CHK_RET(NetworkManager::GetInstance(deviceLogicId_).PsWorkerRaInit(devicePhyId_, loopBackIp_, port));
    workerRaInit_ = true;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitPsRaResource(u32 devLogicId, s32 devicePhyId, const HcclIpAddress &ipAddr, u32 port)
{
    HCCL_INFO("DeInit ipAddr[%s], devicePhyId[%u], deviceLogicId_[%d], port[%u]",
        ipAddr.GetReadableAddress(), devicePhyId, devLogicId, port);
    CHK_RET(NetworkManager::GetInstance(devLogicId).PsWorkerRaDeinit(devicePhyId, ipAddr, port));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitHdcPsRaResource()
{
    std::string hostIp(hostIp_.GetReadableAddress());
    for (const auto &devsInfo : rankDevicePhyIdNicInfoMap_[hostIp]) {
        u32 devLogicId;
        CHK_RET(hrtGetDeviceIndexByPhyId(devsInfo.first, devLogicId));
        HcclNetCloseDev(netDevCtxMap_[devsInfo.second]);
        netDevCtxMap_.erase(devsInfo.second);
        CHK_RET(DeinitPsRaResource(devLogicId, devsInfo.first, devsInfo.second, ranksPort_[realUserRank_]));
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitPsRaResource()
{
    if (psRaInit_) {
        HCCL_INFO("deinit ps ra resource!");
        bool hostIpDeinited = false;
        bool loopBackIpDeinited = false;
        for (const auto &clientInfo : clientsInfo_) {
            if (serverId_ != clientInfo.serverId && !hostIpDeinited) {
                if (isHostUseDevNic_) {
                    // hdc模式下以device为粒度解初始化
                    CHK_RET(DeinitHdcPsRaResource());
                } else {
                    // peer模式下以hostIp出发进行注销
                    CHK_RET(DeinitPsRaResource(deviceLogicId_, devicePhyId_, hostIp_, ranksPort_[userRank_]));
                }
                hostIpDeinited = true;
            } else if (!loopBackIpDeinited) {
                CHK_RET(DeinitPsRaResource(deviceLogicId_, devicePhyId_, loopBackIp_, ranksPort_[userRank_]));
                loopBackIpDeinited = true;
            }
        }
        psRaInit_ = false;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitWorkerRaResource()
{
    if (workerRaInit_) {
        HCCL_INFO("deinit worker ra resource!");
        s32 port = 0;
        if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT) {
            port = HOST_PARA_BASE_PORT + devicePhyId_;
        } else {
            port = devicePhyId_ + GetExternalInputHcclIfBasePort() + HCCL_AISERVER_DEVICE_NUM;
        }

        HCCL_INFO("Deinit loopBackIp[%s], devicePhyId_[%u], deviceLogicId_[%d]!",
            loopBackIp_.GetReadableAddress(), devicePhyId_, deviceLogicId_);
        CHK_RET(NetworkManager::GetInstance(deviceLogicId_).PsWorkerRaDeinit(devicePhyId_, loopBackIp_, port));
        workerRaInit_ = false;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DisablePreResource()
{
    // 查询本rank所在服务器
    auto iterServ = servRankInfo_.find(serverId_);
    bool check = (iterServ == servRankInfo_.end());
    CHK_PRT_RET(check, HCCL_ERROR("[Disable][PreResource]can't find serverId[%s] in server map", serverId_.c_str()),
        HCCL_E_NOT_FOUND);
    HcclResult ret = P2PMgmtPub::DisableP2P(enableP2PDevices_);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[Disable][PreResource]Disable all P2P Failed, deviceLogicId[%d], ret[%u]",
        deviceLogicId_, ret), ret);
    enableP2PDevices_.clear();
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetWorkspaceSubStreamNum(u64 &streamNum, u64 dataSize, HcclCMDType opType)
{
    AlgType algType;

    CHK_RET(GetAlgType(algType, opType));

    auto const algLevel0 = algType.algoLevel0;

    // 根据所用算法，选择所需的从stream数目
    switch (algLevel0) {
        case AlgTypeLevel0::ALG_LEVEL0_NP_MESH:
            streamNum = userRankSize_ / moduleNum_ - HCCL_SUB_STREAM_NP_MESH;
            break;
        case AlgTypeLevel0::ALG_LEVEL0_8P_RING:
            streamNum = HCCL_SUB_STREAM_NUM_8P_RING;
            break;
        case AlgTypeLevel0::ALG_LEVEL0_NP_DOUBLE_RING:
            streamNum = HCCL_SUB_STREAM_NUM_DOUBLE_RING;
            break;
        case AlgTypeLevel0::ALG_LEVEL0_4P_MESH:
            streamNum = HCCL_SUB_STREAM_NUM_4P_MESH;
            break;
        default:
            streamNum = HCCL_SUB_STREAM_NUM_ZERO;
            break;
    }

    auto iter = HCCL_ALGO_LEVEL0_NAME_MAP.find(algLevel0);
    CHK_PRT_RET(iter == HCCL_ALGO_LEVEL0_NAME_MAP.end(),
        HCCL_ERROR("[GetWorkspaceSubStreamNum]level0: algType[%u] is invalid.", algLevel0),
        HCCL_E_INTERNAL);
    HCCL_DEBUG("[GetWorkspaceSubStreamNum]hccl algorithm: In level0, using %s algo, the streamNum is %llu",
        iter->second.c_str(), streamNum);

    u64 sliceNum = CalculatePiplineSliceNum(opType, dataSize, algType, deviceType_, deviceNumPerServer_, serverNum_);
    // 图模式下数据量固定, 按照当前数据量判断是否支持pipline切分并申请从流
    if (implAlg_ != nullptr && sliceNum >= MIN_PIPLINE_SLICE_NUM) {
        streamNum++;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DestroyNetworkResources()
{
    /* 本通信域内只有1个device时，不需要卸载ra资源 */
    if (userRankSize_ <= 1) {
        HCCL_INFO("user rank size <= 1, ra is not needed for single device");
        return HCCL_SUCCESS;
    }

    if (heterogHostNicInitialized_) {
        CHK_RET(DeinitHeterogHostNic());
    }

    // nic的初始化独立调用，在此单独判断是否需要解初始化
    if (nicInitialized_) {
        CHK_RET(DeinitNic());
    }

    if (raResourceInit_ && (static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID)) {
        if (netDevCtxMap_[localVnicIp_] != nullptr) {
            CHK_RET(socketManager_->ServerDeInit(netDevCtxMap_[localVnicIp_], localVnicListenPort_));
            HcclNetCloseDev(netDevCtxMap_[localVnicIp_]);
            netDevCtxMap_.erase(localVnicIp_);
            raResourceInit_ = false;
        }
    }

    if (raResourceInit_) {
        if (static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID) {
            CHK_RET(HcclNetDeInit(NICDeployment::NIC_DEPLOYMENT_DEVICE, devicePhyId_, deviceLogicId_));
        }

        if ((static_cast<s32>(devicePhyId_) != HOST_DEVICE_ID && IsEnableRoce() &&
            nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST) ||
            (Is310PDevice() && nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST) ||
            (!Is310PDevice() && commWorkMode_ != WorkMode::HCCL_MODE_AI_CPU && isHaveCpuRank_)) {
            u32 devicePhyID = (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) ? 0 : devicePhyId_;
            CHK_RET(HcclNetDeInit(NICDeployment::NIC_DEPLOYMENT_HOST, devicePhyID, deviceLogicId_));
        }

        socketManager_ = nullptr;
    }
    return HCCL_SUCCESS;
}

// 用于标识集群中是否存在 910B A+X形态
bool HcclImplBase::IsDiffDeviceModule(const std::vector<RankInfo_t> &rankList) const
{
    bool minDevice = false;
    bool maxDevice = false;
    bool isDiffMeshAggregation = false;
    if (rankList.size() == 0) {
        HCCL_INFO("[IsDiffDeviceModule] rankList.size() is 0");
        return false;
    }
    for (const RankInfo_t &rankInfo : rankList) {
        if (rankInfo.deviceInfo.devicePhyId < DEVICE_PER_MODULE) {
            minDevice = true;
        } else {
            maxDevice = true;
        }
    }
    if (minDevice && maxDevice) {
        isDiffMeshAggregation = true;
    }
    return isDiffMeshAggregation;
}

HcclResult HcclImplBase::SetWorkspaceResource(const std::string &tag, void *memPtr, u64 &maxSize,
    std::vector<rtStream_t> &stream)
{
    return workSpaceRes_->SetWorkspaceResource(tag, memPtr, maxSize, stream);
}

void HcclImplBase::DestroyWorkspaceResource(const std::string &tag)
{
    workSpaceRes_->DestroyWorkspaceResource(tag);
}

HcclResult HcclImplBase::AtomicInitSet()
{
    CHK_PRT_RET(initializedFlag_.test_and_set(), HCCL_ERROR("[HcclImplBase][AtomicInitSet]errNo[0x%016llx] instance "\
        "already been initialized", HCCL_ERROR_CODE(HCCL_E_INTERNAL)), HCCL_E_INTERNAL);
    return HCCL_SUCCESS;
}

void HcclImplBase::AtomicInitClear()
{
    initializedFlag_.clear();
}

u32 HcclImplBase::GetUserRank()
{
    return realUserRank_;
}

u32 HcclImplBase::GetGroupRank()
{
    return userRank_;
}

u32 HcclImplBase::GetRankSize()
{
    return userRankSize_;
}

bool HcclImplBase::GetNicInitialized()
{
    return nicInitialized_;
}

HcclResult HcclImplBase::CheckDeviceType(const DevType deviceType) const
{
    if ((deviceType >= DevType::DEV_TYPE_COUNT) || (deviceType < DevType::DEV_TYPE_910)) {
        HCCL_ERROR("[Check][DeviceType]errNo[0x%016llx] deivce Type[%d] out of range[%d, %d]",
            HCCL_ERROR_CODE(HCCL_E_PARA), deviceType, DevType::DEV_TYPE_910, DevType::DEV_TYPE_NOSOC);
        return HCCL_E_PARA;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckReductionOp(const HcclReduceOp op) const
{
    if ((op >= HCCL_REDUCE_RESERVED) || (op < HCCL_REDUCE_SUM)) {
        HCCL_ERROR("[Check][ReductionOp]errNo[0x%016llx] op:[%d] not supported", HCCL_ERROR_CODE(HCCL_E_PARA), op);
        return HCCL_E_PARA;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckUserRank(const u32 userRank) const
{
    if (userRankSize_ <= userRank) {
        HCCL_ERROR("[Check][UserRank]errNo[0x%016llx] userRank:[%u] is out of range[0 ~ %u]",
            HCCL_ERROR_CODE(HCCL_E_PARA), userRank, userRankSize_);
        return HCCL_E_PARA;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CheckCount(const u64 count) const
{
    if (count > SYS_MAX_COUNT) {
        HCCL_ERROR("[Check][Count]errNo[0x%016llx] count[%llu] is invalid(bigger than MAX count[%llu])",
            HCCL_ERROR_CODE(HCCL_E_PARA), count, SYS_MAX_COUNT);
        return HCCL_E_PARA;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetGroupRanksInfo(const std::vector<u32> &groupRanks, std::vector<RankInfo> &ranksInfo)
{
    ranksInfo.clear();
    std::vector<RankInfo> tmpRankInfoList;
    tmpRankInfoList.assign(rankInfoList_.begin(), rankInfoList_.end());

    for (u32 index = 0; index < groupRanks.size(); index++) {
        if (tmpRankInfoList.size() <= groupRanks[index]) {
            HCCL_ERROR("[Get][GroupRanksInfo]errNo[0x%016llx] groupRanks[%u]=[%u], >= rankinfolist size[%zu]",
                HCCL_ERROR_CODE(HCCL_E_PARA), index, groupRanks[index], tmpRankInfoList.size());
            return HCCL_E_PARA;
        }
        tmpRankInfoList[groupRanks[index]].userRank = index;
        ranksInfo.push_back(tmpRankInfoList[groupRanks[index]]);
    }

    // 按rank id从小到大的顺序返回
    std::sort(ranksInfo.begin(), ranksInfo.end(), CompareWithUserRank);

    for (u32 index = 0; index < ranksInfo.size(); ++index) {
        if (index != ranksInfo[index].userRank) {
            HCCL_ERROR("[Get][GroupRanksInfo]errNo[0x%016llx] index[%u] !=  user rank[%u]",
                HCCL_ERROR_CODE(HCCL_E_PARA), index, ranksInfo[index].userRank);
            return HCCL_E_PARA;
        }
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetGroupCommonData(WorldGroupInfo &groupCommonData) const
{
    groupCommonData.inlineReduceSwitchOn = inlineReduceSwitchOn_;
    groupCommonData.deviceType = deviceType_;
    groupCommonData.deviceLogicId = deviceLogicId_;
    groupCommonData.profilingInitiated = profilingInitiated_;
    groupCommonData.serverId = serverId_;
    groupCommonData.phyIdNicInfoMap = rankDevicePhyIdNicInfoMap_;
    groupCommonData.worldRankInfoList = rankInfoList_;
    groupCommonData.ranksPort = ranksPort_;

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetWorkspaceMemSize(const std::string &opType, u64 count, HcclDataType dataType,
    u32 &rankSize, u64 &memSize, DevType &deviceType) const
{
    return workSpaceRes_->GetWorkspaceMemSize(opType, count, dataType, rankSize, memSize, deviceType);
}

HcclResult HcclImplBase::InitProfiling()
{
    if (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
        return HCCL_SUCCESS;
    }
    CHK_PRT_RET(profilingInitiated_, HCCL_DEBUG("Profiling plugin has already been Initiated."), HCCL_SUCCESS);

    if (profilingMode_ != HcomProfilingMode::PROFILING_OPEN && GetExternalInputProfilingMode()) {
        profilingMode_ = HcomProfilingMode::PROFILING_OPEN;
        profilingOption_ = GetExternalInputProfilingOption();
    }
    HCCL_INFO("profiling config information:options[%s], mode[%d]", profilingOption_.c_str(), profilingMode_);

    // profilingInitiated_会广播给所有子通信域，用于避免taskInfoSaver的重复初始化
    profilingInitiated_ = true;
    // isExecuteProfilingInit_用于记录本impl是否执行了taskInfoSaver的初始化，用于进行对应的释放
    isExecuteProfilingInit_ = true;
    return HCCL_SUCCESS;
}

static bool g_profilerInit = false;

HcclResult HcclImplBase::InitESProfiler()
{
    if (g_profilerInit) {
        return HCCL_SUCCESS;
    }

    CHK_RET(DlProfFunction::GetInstance().DlProfFunctionInit());
    CHK_RET(DlRtFunction::GetInstance().DlRtFunctionInit());
    g_profilerInit = true;
    rtProfCtrlHandle callback = EsCommandHandle;

    HcclResult ret = hrtProfRegisterCtrlCallback(HCCL, callback);
    CHK_PRT_RET((ret != HCCL_SUCCESS), HCCL_ERROR("[BASE][InitESProfiler]Register CtrlCallBack failed."),
        HCCL_E_PARA);
    HCCL_INFO("[Profiling][InitESProfiler]Register CtrlCallBack success");

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitProfiling()
{
    CHK_PRT_RET(!profilingInitiated_, HCCL_DEBUG("Profiling plugin has not been Initiated"), HCCL_SUCCESS);
    profilingInitiated_ = false;
    HCCL_INFO("Profiling is deinitiated.");
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::RegistTaskExceptionHandler() const
{
    if ((static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) || (isHeterogComm_)) {
        return HCCL_SUCCESS;
    }
    CHK_RET(TaskExceptionHandler::Init());

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::UnRegistTaskExceptionHandler() const
{
    if ((static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) || (isHeterogComm_)) {
        return HCCL_SUCCESS;
    }
    CHK_RET(TaskExceptionHandler::DeInit());

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetInCCLbuffer(void* &buffer, u64 &size)
{
    return cclBufferManager_.GetInCCLbuffer(buffer, size);
}

HcclResult HcclImplBase::GetOutCCLbuffer(void* &buffer, u64 &size)
{
    return cclBufferManager_.GetOutCCLbuffer(buffer, size);
}

void HcclImplBase::ReleaseCommCCLbuffer()
{
    cclBufferManager_.ReleaseCommCCLbuffer();
}

HcclResult HcclImplBase::ReleaseCommInfos()
{
    if (implAlg_ != nullptr) {
        return implAlg_->ReleaseCommInfos();
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitProfiler()
{
    if (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
        return HCCL_SUCCESS;
    }

    profilerManager_.reset(new (std::nothrow) ProfilerManager(devicePhyId_, deviceLogicId_, realUserRank_));
    CHK_SMART_PTR_NULL(profilerManager_);
    HcclResult ret = profilerManager_->InitProfiler();
    CHK_PRT_RET((ret != HCCL_SUCCESS), HCCL_ERROR("[BASE][InitProfiler]profilerManager_ InitProfiler failed."),
        HCCL_E_PARA);

    HCCL_INFO("[BASE][InitProfiler]Register CtrlCallBack success");

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CreateCommCCLbuffer()
{
    return cclBufferManager_.CreateCommCCLbuffer();
}

HcclResult HcclImplBase::InitCCLbuffer(u64 inCCLbufferSize, u64 outCCLbufferSize)
{
    return cclBufferManager_.InitCCLbuffer(inCCLbufferSize, outCCLbufferSize);
}

HcclResult HcclImplBase::InitNic()
{
    if (!GetExternalInputIntraRoceSwitch() && servRankInfo_.size() == 1 && isDiffDeviceModule_) {
        return HCCL_SUCCESS;
    }

    u32 port = GetLocalNicPort();
    if (nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
        u32 nicNum = devIpAddr_.size();
        for (u32 i = 0; i < nicNum; i++) {
            if (devIpAddr_[i].IsInvalid()) {
                HCCL_INFO("[Init][Nic]nic num[%u] deviceip is invalid, total nicNum[%u]", i, nicNum);
                continue;
            }
            isUsedRdmaLevel0_ = IsSupportEnableRoce();
            HcclNetDevCtx nicPortCtx;
            CHK_RET(HcclNetOpenDev(&nicPortCtx, NicType::DEVICE_NIC_TYPE, devicePhyId_, deviceLogicId_, devIpAddr_[i]));
            netDevCtxMap_.insert(std::make_pair(devIpAddr_[i], nicPortCtx));
            CHK_RET(socketManager_->ServerInit(nicPortCtx, port));
        }
    } else if (nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST) {
        CHK_PRT_RET((hostIp_.IsInvalid()), HCCL_ERROR("[Init][Nic] host ip is invalid when NIC "
        "deployment is host. "), HCCL_E_PARA);
        isUsedRdmaLevel0_ = IsSupportEnableRoce();
        u32 devicePhyID = (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) ? 0 : devicePhyId_;
        HCCL_INFO("[Init][Nic], hostPort[%u], devicePhyID[%u]", port, devicePhyID);
        HcclNetDevCtx hostnicPortCtx;
        CHK_RET(HcclNetOpenDev(&hostnicPortCtx, NicType::HOST_NIC_TYPE, devicePhyID, deviceLogicId_, hostIp_));
        netDevCtxMap_.insert(std::make_pair(hostIp_, hostnicPortCtx));
        CHK_RET(socketManager_->ServerInit(hostnicPortCtx, port));
    } else {
        HCCL_ERROR("[Init][Nic]nic deployment[%d] is not supported", nicDeployment_);
        return HCCL_E_PARA;
    }
    nicInitialized_ = true;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitNic()
{
    u32 port = GetLocalNicPort();

    if (nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
        u32 nicNum = devIpAddr_.size();
        for (u32 i = 0; i < nicNum; i++) {
            if (devIpAddr_[i].IsInvalid()) {
                continue;
            }

            CHK_RET(socketManager_->ServerDeInit(netDevCtxMap_[devIpAddr_[i]], port));
            HcclNetCloseDev(netDevCtxMap_[devIpAddr_[i]]);
            netDevCtxMap_.erase(devIpAddr_[i]);
        }
    } else if (nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST) {
        CHK_PRT_RET((hostIp_.IsInvalid()), HCCL_ERROR("[DeInit][Nic] host ip is invalid when NIC "
        "deployment is host. "), HCCL_E_PARA);
        HCCL_INFO("[Deinit][Nic], hostPort[%u]", port);
        if (netDevCtxMap_.find(hostIp_) != netDevCtxMap_.end()) {
            CHK_RET(socketManager_->ServerDeInit(netDevCtxMap_[hostIp_], port));
            HcclNetCloseDev(netDevCtxMap_[hostIp_]);
            netDevCtxMap_.erase(hostIp_);
        }
    } else {
        HCCL_ERROR("[Deinit][Nic]nic deployment[%d] is not supported", nicDeployment_);
        return HCCL_E_PARA;
    }
    nicInitialized_ = false;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitHeterogHostNic(void)
{
    bool needHostNic = false;
    for (u32 i = 0; i < rankInfoList_.size(); i++) {
        if (rankInfoList_[i].devicePhyId == HOST_DEVICE_ID) {
            needHostNic = true;
            break;
        }
    }

    if (needHostNic) {
        HCCL_INFO("[Init][HostNic], hostIp[%s] hostPort[%u] devLogicId[%d]", hostIp_.GetReadableAddress(), hostPort_,
            deviceLogicId_);

        if (netDevCtxMap_.find(hostIp_) == netDevCtxMap_.end()) {
            HcclNetDevCtx hostnicPortCtx;
            CHK_RET(HcclNetOpenDev(&hostnicPortCtx, NicType::HOST_NIC_TYPE, devicePhyId_, deviceLogicId_, hostIp_));
            netDevCtxMap_.insert(std::make_pair(hostIp_, hostnicPortCtx));
            CHK_RET(socketManager_->ServerInit(hostnicPortCtx, hostPort_));
        }

        if (!(hostIp_ == loopBackIp_)) {
            HcclNetDevCtx loopbackPortCtx;
            CHK_RET(
                HcclNetOpenDev(&loopbackPortCtx, NicType::HOST_NIC_TYPE, devicePhyId_, deviceLogicId_, loopBackIp_));
            netDevCtxMap_.insert(std::make_pair(loopBackIp_, loopbackPortCtx));
            CHK_RET(socketManager_->ServerInit(loopbackPortCtx, hostPort_));
        }
    }
    heterogHostNicInitialized_ = true;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeinitHeterogHostNic()
{
    if (!heterogHostNicInitialized_) {
        return HCCL_SUCCESS;
    }
    bool needHostNic = false;
    for (u32 i = 0; i < rankInfoList_.size(); i++) {
        if (rankInfoList_[i].devicePhyId == HOST_DEVICE_ID) {
            needHostNic = true;
            break;
        }
    }

    if (needHostNic) {
        HCCL_INFO("[Deinit][HostNic], hostIp[%s] hostPort[%u]", hostIp_.GetReadableAddress(), hostPort_);
        if (!(hostIp_ == loopBackIp_)) {
            CHK_RET(socketManager_->ServerDeInit(netDevCtxMap_[loopBackIp_], hostPort_));
            HcclNetCloseDev(netDevCtxMap_[loopBackIp_]);
            netDevCtxMap_.erase(loopBackIp_);
        }

        if (netDevCtxMap_.find(hostIp_) != netDevCtxMap_.end()) {
            CHK_RET(socketManager_->ServerDeInit(netDevCtxMap_[hostIp_], hostPort_));
            HcclNetCloseDev(netDevCtxMap_[hostIp_]);
            netDevCtxMap_.erase(hostIp_);
        }
    }
    heterogHostNicInitialized_ = false;
    return HCCL_SUCCESS;
}

void HcclImplBase::DestroyHeterogTransport()
{
    std::unique_lock<SpinMutex> transportMapLock(transportMapSpinMutex_);
    transportStorage_.clear();
    return;
}

HcclResult HcclImplBase::SetGlobalWorkSpace(std::vector<void *> &globalWorkSpaceAddr)
{
    CHK_RET(HcclSetGlobalWorkSpace(dispatcher_, globalWorkSpaceAddr));
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetandClearOverFlowTasks(std::vector<HcclDumpInfo> &hcclDumpInfo)
{
    if (profilerManager_ != nullptr) {
        CHK_RET(profilerManager_->GetandClearOverFlowTasks(hcclDumpInfo));
    } else {
        HCCL_WARNING("[impl][GetDumpTask] profilerManager_ not set");
    }
    return HCCL_SUCCESS;
}

bool HcclImplBase::IsOverFlowInfnanMode() const
{
    rtFloatOverflowMode_t floatOverflowMode = RT_OVERFLOW_MODE_UNDEF;
    HcclResult ret = hrtGetDeviceSatMode(&floatOverflowMode);
    if (ret != HCCL_SUCCESS) {
        HCCL_WARNING("[impl][IsOverFlowInfnanMode] GetDeviceSatMode failed");
    }
    return (!GetExternalInputHcclDumpDebug()) && (floatOverflowMode == RT_OVERFLOW_MODE_INFNAN);
}

HcclResult HcclImplBase::GetDeviceId(s32 &deviceId) const
{
    deviceId = deviceLogicId_;
    return HCCL_SUCCESS;
}

bool HcclImplBase::GetBoardVersion(std::vector<RoleTableNodeInfo> &nodes_)
{
    if (ifBoardVersionInit_) {
        return isBoardVersion_;
    }

    for (const auto &node : nodes_) {
        if (serverId_ == node.serverId) {
            isBoardVersion_ = true;
            break;
        }
    }
    ifBoardVersionInit_ = true;

    return isBoardVersion_;
}

HcclResult HcclImplBase::BuildEsWorkerTransportLinks(const s32 &tag)
{
    uint64_t beginTime = hrtMsprofSysCycleTime();
    for (const auto &node : clientsInfo_) {
        HCCL_INFO("clientsInfo_ size[%u], rankId[%u], userRank_[%u].", clientsInfo_.size(), node.rankId, userRank_);
        if (node.rankId == userRank_) {
            // 和所有的servers建链
            CHK_RET(BuildTransport(node.port, node.hostPort, tag, serversInfo_));
            break;
        }
    }

    auto &profilingManager = ProfilingManager::Instance();
    CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_BUILD_CS_TRANSPORT));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitPsRdmaHandle(u32 devLogicId, s32 devicePhyId, const HcclIpAddress &ipAddr,
    bool disabledLiteThread, bool enable910ALite)
{
    HCCL_INFO("devicePhyId[%d] logicId[%d] ipAddr[%s], disabledLiteThread[%u], enable910ALite[%u].",
        devicePhyId, devLogicId, ipAddr.GetReadableAddress(), disabledLiteThread, enable910ALite);
    CHK_RET(NetworkManager::GetInstance(devLogicId).InitRdmaHandle(devicePhyId, ipAddr,
        disabledLiteThread, enable910ALite));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitHdcPsRdmaHandle()
{
    // hdc模式下，rdma通信ps使用device网卡，轮询每个device网卡初始化rdma
    std::string hostIp(hostIp_.GetReadableAddress());
    for (const auto &devsInfo : rankDevicePhyIdNicInfoMap_[hostIp]) {
        u32 devLogicId;
        CHK_RET(hrtGetDeviceIndexByPhyId(devsInfo.first, devLogicId));
        // 910A环境下ES业务需要放开910A环境的lite模式
        CHK_RET(InitPsRdmaHandle(devLogicId, devsInfo.first, devsInfo.second, true, true));
    }
    return HCCL_SUCCESS;
}

bool HcclImplBase::IsNeedGlobalMrManager()
{
    // 非hdc模式或者910B环境的ps需要维护全局的MrManager(以减少内存开销)
    bool isLocalDevNicInPlane;
    CHK_RET(IsLocalDevNicInPlane(isLocalDevNicInPlane));
    return (!isHostUseDevNic_ || (!isLocalDevNicInPlane && deviceLogicId_ == HOST_DEVICE_ID));
}

HcclResult HcclImplBase::InitPsRdmaHandle()
{
    if (hostRdmaInitFlag_) {
        return HCCL_SUCCESS;
    }

    for (const auto &clientInfo : clientsInfo_) {
        if (serverId_ != clientInfo.serverId) {
            if (isHostUseDevNic_) {
                CHK_RET(InitHdcPsRdmaHandle());
            } else {
                // peer模式以hostIp创建RDMA通信句柄
                CHK_RET(InitPsRdmaHandle(deviceLogicId_, devicePhyId_, hostIp_));
            }

            if (IsNeedGlobalMrManager()) {
                // ps侧使用全局的MrManager时走进来
                CHK_RET(MrManagerInit());
            }
            CHK_RET(InitMemBlocksAndRecvWrMem());
            hostRdmaInitFlag_ = true;
            break;
        }
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::TransportInitThread(std::unique_ptr<TransportHeterog> &transportPtr, HcclResult &ret)
{
    HCCL_INFO("[TransportInitThread] entry.");

    ret = transportPtr->Init();
    CHK_RET(ret);
    HCCL_DEBUG("BuildShmTransport: SHM transport connect success!.");
    HCCL_INFO("[TransportInitThread] exit");
    return HCCL_SUCCESS;
}

// 判断本端网卡是否分平面
HcclResult HcclImplBase::IsLocalDevNicInPlane(bool &isLocalDevNicInPlane)
{
    // 只有AI-Server的910A环境分平面
    if (!isHostUseDevNic_) {
        // 非Ai-server不分平面
        isLocalDevNicInPlane = false;
        return HCCL_SUCCESS;
    }
    DevType devType;
    CHK_RET(hrtHalGetDeviceType(DEVICE_LOGIC_ID_ZERO, devType));
    // 非80环境下基于不分平面的网卡进行注册和建链，以减少注册开销
    isLocalDevNicInPlane = (devType == DevType::DEV_TYPE_910) ? true : false;

    return HCCL_SUCCESS;
}

// 判断对端网卡是否分平面
HcclResult HcclImplBase::IsRemoteDevNicInPlane(bool &isRemoteDevNicInPlane)
{
    // 只有AI-Server的910A环境分平面
    if (!GetRemoteIsHdc()) {
        // 对端非Ai-server不分平面
        isRemoteDevNicInPlane = false;
        return HCCL_SUCCESS;
    }
    // 目前没有办法获取对端的芯片类型，默认在对端的不是通用服务器或者51环境的前提下，对端的芯片类型与本端相同
    DevType devType;
    CHK_RET(hrtHalGetDeviceType(DEVICE_LOGIC_ID_ZERO, devType));
    // 非80环境下基于不分平面的网卡进行注册和建链，以减少注册开销
    isRemoteDevNicInPlane = (devType == DevType::DEV_TYPE_910) ? true : false;

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetDevIP(const u32 &psRankId, const HcclIpAddress& hostIp,
    u32 &devicePhyId, HcclIpAddress& NicIp)
{
    // 根据RankId在hostIp所在server内按特定规则分配device网卡，用于Aiserver场景网络不分平面的server间通信
    /*
    网卡不分平面时，为每个ps分配一个device网卡以进行rdma通信
    1. 对目标server内的ps按照rankId进行排序，得到排序系数idx
    2. 按照排名为在rankDevicePhyIdNicInfoMap_内选择对应hostIp下的第idx个键值对作为分配给当前ps的device网卡
    3. 若当前idx超出网卡列表长度时对idx做关于长度的模运算
    该处理可避免同server内的psRankId不连续时可能出现的多个ps分到同一张device网卡的潜在问题
    使用该函数之前须通过GetDeviceNicInfo向rankDevicePhyIdNicInfoMap_添加网卡信息
    */
    set<u32> psRankIdSet;
    for (auto &serverInfo : serversInfo_) {
        if (hostIp == serverInfo.hostIp) {
            psRankIdSet.insert(serverInfo.rankId);
        }
    }
    auto it = find(psRankIdSet.begin(), psRankIdSet.end(), psRankId);
    CHK_PRT_RET(it == psRankIdSet.end(), HCCL_ERROR("psRankId[%u] not found in psRankIdSet", psRankId), HCCL_E_PARA);
    if (rankDevicePhyIdNicInfoMap_.find(hostIp.GetReadableAddress()) == rankDevicePhyIdNicInfoMap_.end()) {
        HCCL_RUN_WARNING("Get available device nic info map for hostIp[%s] fail,"
            "maybe node is general server, use host ipAddr", hostIp.GetReadableAddress());
        return HCCL_SUCCESS;
    }
    int availNicIdx = distance(psRankIdSet.begin(), it) %
        rankDevicePhyIdNicInfoMap_[hostIp.GetReadableAddress()].size();
    auto availNicIter = next(rankDevicePhyIdNicInfoMap_[hostIp.GetReadableAddress()].begin(), availNicIdx);
    CHK_PRT_RET(availNicIter == rankDevicePhyIdNicInfoMap_[hostIp.GetReadableAddress()].end(),
        HCCL_ERROR("Get device nic info for hostIp[%s] fail",
        hostIp.GetReadableAddress()), HCCL_E_PARA);
    devicePhyId = availNicIter->first;
    NicIp = availNicIter->second;
    HCCL_INFO("Get device nic[%s] devicePhyId[%u] for hostIp[%s], psRankId[%u] success!", NicIp.GetReadableAddress(),
        devicePhyId, hostIp.GetReadableAddress(), psRankId);
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::WaitBuildLinkComplete(const u32 remoteRank,
    const std::unique_ptr<TransportHeterog>& transportPtr)
{
    HcclResult ret = transportPtr->WaitBuildLinkComplete();
    if (ret == HCCL_E_TIMEOUT) {
        HCCL_ERROR("[WaitBuildLinkComplete]localRank[%u], remoteRank[%u]", userRank_, remoteRank);
    }
    return ret;
}

HcclResult HcclImplBase::BuildRdmaTransport(const u32 &nodePort, const RoleTableNodeInfo& node,
    MachinePara& machinePara, std::unique_ptr<TransportHeterog>& transportPtr)
{
    CHK_PRT_RET(node.port > MAX_PORT_ID || node.port < MIN_PORT_ID,
        HCCL_ERROR("[NetworkManager][HeterogStartListen] port error[%u]", node.port), HCCL_E_NETWORK);

    HcclIpAddress selfIp = devIpAddr_[0];
    u32 selfPort = nodePort;
    HcclIpAddress peerIp = node.ipAddr;
    u32 peerPort = node.port;
    machinePara.nicDeploy = NICDeployment::NIC_DEPLOYMENT_DEVICE;
    std::chrono::milliseconds kdefaultTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
    transportResourceInfo_.isESMode = true;

    if (isHostUseDevNic_) {
        u32 targetDevicePhyId;
        u32 devLogicId;
        DevType devType;
        transportResourceInfo_.isHdcMode = true;
        transportResourceInfo_.deviceLogicId = deviceLogicId_;
        CHK_PRT(GetMemBlockNum(devicePhyId_, transportResourceInfo_.memBlockNum));

        if (role_ == SERVER_ROLE_SOCKET) {
            // 根据本端网卡是否分平面为当前ps选择device网卡
            bool isLocalDevNicInPlane;
            CHK_RET(IsLocalDevNicInPlane(isLocalDevNicInPlane));
            if (!isLocalDevNicInPlane) {
                CHK_RET(GetDevIP(userRank_, hostIp_, targetDevicePhyId, selfIp));
            } else {
                targetDevicePhyId = node.devicePhyId;
                CHK_RET(GetDevIP(hostIp_, targetDevicePhyId, selfIp));
            }

            // hdc模式下ps进行RDMA通信使用单算子模式
            CHK_RET(hrtGetDeviceIndexByPhyId(targetDevicePhyId, devLogicId));
            CHK_RET(hrtHalGetDeviceType(devLogicId, devType));
            if (devType == DevType::DEV_TYPE_910B) {
                HCCL_DEBUG("qpmode is OPBASE_QP_MODE_EXT");
                transportResourceInfo_.qpMode = OPBASE_QP_MODE_EXT;
            } else {
                transportResourceInfo_.qpMode = OPBASE_QP_MODE;
            }
            u32 linksMaxBlocks = transportResourceInfo_.memBlockNum / (clientsInfo_.size() * ES_STREAM_LIMIT * OP_NUM);
            transportResourceInfo_.hdcHostWqeBatchNum = transportResourceInfo_.hdcHostWqeBatchNum < linksMaxBlocks ?
                transportResourceInfo_.hdcHostWqeBatchNum : linksMaxBlocks;
            if (transportResourceInfo_.hdcHostWqeBatchNum < OP_NUM) {
                HCCL_WARNING("hdcHostWqeBatchNum Must be greater than 2");
                transportResourceInfo_.hdcHostWqeBatchNum = OP_NUM;
            }
        } else {
            // 根据对端网卡是否分平面获取对端ps使用的device网卡ip
            targetDevicePhyId = devicePhyId_;
            bool isRemoteDevNicInPlane;
            CHK_RET(IsRemoteDevNicInPlane(isRemoteDevNicInPlane));
            if (!isRemoteDevNicInPlane) {
                // worker侧不关心对端使用的网卡的物理Id，这里只是用作下面接口的出参
                u32 remDevicePhyId;
                CHK_RET(GetDevIP(node.rankId, node.hostIp, remDevicePhyId, peerIp));
            } else {
                CHK_RET(GetDevIP(node.hostIp, targetDevicePhyId, peerIp));
            }

            // hdc模式下worker进行RDMA通信使用下沉模式
            CHK_RET(hrtGetDeviceIndexByPhyId(targetDevicePhyId, devLogicId));
            CHK_RET(hrtHalGetDeviceType(devLogicId, devType));
            if (devType == DevType::DEV_TYPE_910B) {
                HCCL_DEBUG("qpmode is OFFLINE_QP_MODE_EXT");
                s32 qpMode = (GetWorkflowMode() ==
                    HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? OPBASE_QP_MODE_EXT : OFFLINE_QP_MODE_EXT);
                transportResourceInfo_.qpMode = qpMode;
            } else {
                s32 qpMode = (GetWorkflowMode() ==
                    HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ? OPBASE_QP_MODE : OFFLINE_QP_MODE);
                transportResourceInfo_.qpMode = qpMode;
            }
        }

        HCCL_INFO("selfIp[%s] peerIp[%s] selfPort[%u] peerPort[%u].", selfIp.GetReadableAddress(),
            peerIp.GetReadableAddress(), selfPort, peerPort);

        // 下沉rdma模式(包括下沉和单算子模式)使用TransportHeterogRoce并推动建链完成
        transportPtr.reset(new (std::nothrow)
            TransportHeterogRoce(machinePara.tag, selfIp, peerIp, peerPort, selfPort, transportResourceInfo_));
        CHK_PRT_RET(!transportPtr, HCCL_ERROR("[Init][Transport]In create link, new link failed."), HCCL_E_PTR);
        CHK_RET(transportPtr->SetDeviceIndex(devLogicId));
        CHK_RET(transportPtr->Init());

        HCCL_INFO("BuildRdmaTransport: HostUseDevNic RDMA transport connection created! remote rankId[%d].",
            node.rankId);

        // PS当前需要阻塞建链
        if (role_ == SERVER_ROLE_SOCKET) {
            CHK_RET(WaitBuildLinkComplete(node.rankId, transportPtr));
            HCCL_INFO("BuildRdmaTransport: HostUseDevNic RDMA transport connect success! remote rankId[%d].",
                node.rankId);
        }
    } else {
        HCCL_INFO("selfIp[%s] peerIp[%s] selfPort[%u] peerPort[%u].", selfIp.GetReadableAddress(),
            peerIp.GetReadableAddress(), selfPort, peerPort);
        // 通用服务器做ps时使用TransportHeterogRoce
        if ((role_ == SERVER_ROLE_SOCKET) && GetRemoteIsHdc()) {
            transportResourceInfo_.isHdcMode = false;
            transportResourceInfo_.deviceLogicId = deviceLogicId_;
            transportResourceInfo_.qpMode = NORMAL_QP_MODE;
            transportResourceInfo_.remoteIsHdc = true;

            transportPtr.reset(new (std::nothrow)
            TransportHeterogRoce(machinePara.tag, selfIp, peerIp, peerPort, selfPort, transportResourceInfo_));
            CHK_PRT_RET(!transportPtr, HCCL_ERROR("[Init][Transport]In create link, new link failed."), HCCL_E_PTR);
            CHK_RET(transportPtr->SetDeviceIndex(deviceLogicId_));
            CHK_RET(transportPtr->Init());
            CHK_RET(WaitBuildLinkComplete(node.rankId, transportPtr));
            HCCL_INFO("BuildRdmaTransport: HostUseDevNic RDMA transport connect success! remote rankId[%d].",
                node.rankId);

        } else {
            // 非HDC模式走非下沉模式的RDMA
            transportPtr.reset(new (std::nothrow)
                TransportRoce(dispatcher_, notifyPool_, machinePara, kdefaultTimeout, selfIp, peerIp,
                selfPort, peerPort, transportResourceInfo_));
            CHK_PRT_RET(!transportPtr, HCCL_ERROR("[Init][Transport]In create link, new link failed."),
                HCCL_E_PTR);
            CHK_RET(transportPtr->Init());
        }

        HCCL_INFO("BuildRdmaTransport: peer RDMA transport connection created! remote rankId[%d].", node.rankId);
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::BuildShmTransport(const u32 &hostPort, const s32 &tag, const RoleTableNodeInfo& node,
    MachinePara& machinePara, std::unique_ptr<TransportHeterog>& transportPtr)
{
    HCCL_DEBUG("transport mode is SHM, remote rankId[%d].", node.rankId);

    if (role_ == CLIENT_ROLE_SOCKET) {
        unique_lock<std::mutex> lockTransferMemsMap(transferMemsMapMutex_);
        machinePara.inputMem = transferMemsMap_[tag][0];
        machinePara.outputMem = transferMemsMap_[tag][1];
        lockTransferMemsMap.unlock();

        HCCL_DEBUG("BuildShmTransport: key mem[%p] size %llu, value mem[%p] size %llu", machinePara.inputMem.ptr(),
            machinePara.inputMem.size(), machinePara.outputMem.ptr(), machinePara.outputMem.size());
    }
    std::chrono::milliseconds kdefaultTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
    HcclIpAddress selfIp = loopBackIp_;
    u32 selfPort = hostPort;
    HcclIpAddress peerIp = loopBackIp_;
    u32 peerPort = node.hostPort;
    u32 remoteDevLogicId = node.devicePhyId;
    if (node.devicePhyId != HOST_DEVICE_ID) {
        CHK_RET(hrtGetDeviceIndexByPhyId(node.devicePhyId, remoteDevLogicId));
        machinePara.remoteDeviceId = remoteDevLogicId;
        HCCL_DEBUG("BuildShmTransport: logicDevId=%u.", machinePara.remoteDeviceId);
    }

    HcclRtContext ctx = nullptr;
    CHK_RET(hrtCtxGetCurrent(&ctx));
    ctx = ((role_ == SERVER_ROLE_SOCKET) ? rtCtxMap_[remoteDevLogicId] : ctx);
    HCCL_INFO("selfIp[%s] peerIp[%s] selfPort[%u] peerPort[%u], node.devicePhyId[%d], remoteDevLogicId[%d]"
        " rtCtx[%p].",
        selfIp.GetReadableAddress(), peerIp.GetReadableAddress(), selfPort, peerPort, node.devicePhyId,
        remoteDevLogicId, ctx);
    HCCL_INFO("serverId_[%s], node.serverId[%s].", serverId_.c_str(), node.serverId.c_str());
    transportPtr.reset(new (std::nothrow)
        TransportShmEvent(dispatcher_, notifyPool_, machinePara, kdefaultTimeout, selfIp, peerIp, peerPort,
        selfPort, deviceLogicId_, role_, transportResourceInfo_, ctx));
    CHK_PRT_RET(!transportPtr, HCCL_ERROR("[Init][Transport]In create link, new link failed."),
        HCCL_E_PTR);
    std::unique_ptr<std::thread> transportInitThreadPtr;
    HcclResult ret;
    transportInitThreadPtr.reset(new (std::nothrow) std::thread(&HcclImplBase::TransportInitThread, this,
        std::ref(transportPtr), std::ref(ret)));
    CHK_PRT_RET(!transportInitThreadPtr, HCCL_ERROR("[BuildShmTransport]threads failed "),
        HCCL_E_INTERNAL);

    if (transportInitThreadPtr->joinable()) {
        transportInitThreadPtr->join();  // 等待线程执行完毕
    }
    CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[BuildShmTransport]init failed, ret[%u]", ret), ret);

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::BuildTcpTransport(const u32 &nodePort, const s32 &tag, const RoleTableNodeInfo& node,
    std::unique_ptr<TransportHeterog>& transportPtr)
{
    HcclIpAddress selfIp = devIpAddr_[0];
    u32 selfPort = nodePort;
    HcclIpAddress peerIp = node.ipAddr;
    u32 peerPort = node.port;
    HCCL_INFO("selfIp[%s] peerIp[%s] selfPort[%u] peerPort[%u].", selfIp.GetReadableAddress(),
        peerIp.GetReadableAddress(), selfPort, peerPort);

    TcpSendThreadPool::GetSendPoolInstance()->Init(devicePhyId_);
    string transTag = to_string(tag);
    transportPtr.reset(new (std::nothrow)TransportHeterogEventTcp(transTag, selfIp, peerIp, peerPort,
        selfPort, devicePhyId_, transportResourceInfo_));
    CHK_PRT_RET(!transportPtr, HCCL_ERROR("[Init][Transport]In create link, new link failed."),
        HCCL_E_PTR);
    CHK_RET(transportPtr->Init(userRank_, node.rankId));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetDevIP(const HcclIpAddress& hostIp, const u32& devicePhyId,
    HcclIpAddress& ip)
{
    // 根据物理Id在hostIp所在server内获取对应device网卡的ip；用于Aiserver场景网络分平面的server间通信
    CHK_PRT_RET(rankDevicePhyIdNicInfoMap_.find(hostIp.GetReadableAddress()) == rankDevicePhyIdNicInfoMap_.end() ||
        rankDevicePhyIdNicInfoMap_[hostIp.GetReadableAddress()].find(devicePhyId) ==
        rankDevicePhyIdNicInfoMap_[hostIp.GetReadableAddress()].end(),
        HCCL_ERROR("Get available device nic info fail, hostIp[%s] devicePhyId[%u]",
        hostIp.GetReadableAddress(), devicePhyId), HCCL_E_PARA);
    ip = rankDevicePhyIdNicInfoMap_[hostIp.GetReadableAddress()][devicePhyId];
    HCCL_INFO("Get available device nic info success, hostIp[%s] devicePhyId[%u] device ip[%s]",
        hostIp.GetReadableAddress(), devicePhyId, ip.GetReadableAddress());

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::BuildEsInnerTransport(MachinePara &machinePara, const u32 &hostPort, const s32 &tag,
    const RoleTableNodeInfo& node, std::unique_ptr<TransportHeterog> &transportPtr)
{
    HCCL_INFO("Build Inner Transport serverId_[%s], localIp[%s], localRank[%u], node.serverId[%s], nodeIp[%s]"
        "remoteRank[%u]", serverId_.c_str(), devIpAddr_[0].GetReadableAddress(), userRank_,
        node.serverId.c_str(), node.ipAddr.GetReadableAddress(), node.rankId);
    machinePara.remoteUserrank = node.rankId;
    machinePara.remoteWorldRank = node.rankId;
    CHK_RET(BuildShmTransport(hostPort, tag, node, machinePara, transportPtr));
    if (role_ == SERVER_ROLE_SOCKET) {
        TransportShmEvent *ptr = dynamic_cast<TransportShmEvent *>(transportPtr.get());
        HCCL_INFO("tag[%d], node.rankId[%u], serverId[%s] GetScratchMem", tag, node.rankId, serverId_.c_str());
        ptr->GetScratchMem(tagRankTransMemsMap_[tag][node.rankId]);
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::BuildEsInterTransport(MachinePara &machinePara, const u32 &nodePort, const s32 &tag,
    const RoleTableNodeInfo& node, std::unique_ptr<TransportHeterog> &transportPtr)
{
    HCCL_INFO("Build Inter Transport serverId_[%s], localIp[%s], localRank[%u], node.serverId[%s], nodeIp[%s]"
        "remoteRank[%u]", serverId_.c_str(), devIpAddr_[0].GetReadableAddress(), userRank_,
        node.serverId.c_str(), node.ipAddr.GetReadableAddress(), node.rankId);
    machinePara.remoteUserrank = node.rankId;
    machinePara.remoteWorldRank = node.rankId;

    if (GetExternalInputHcclIsTcpMode()) {
        CHK_RET(BuildTcpTransport(nodePort, tag, node, transportPtr));
    } else {
        CHK_RET(BuildRdmaTransport(nodePort, node, machinePara, transportPtr));
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CreateAndSaveTransport(MachinePara &machinePara, const u32 &nodePort, const u32 &hostPort,
    const s32 &tag, const RoleTableNodeInfo& node, bool isServerInner)
{
    HCCL_INFO("CreateAndSaveTransport start isServerInner[%d]", isServerInner);
    HcclResult ret;
    {
        lock_guard<shared_timed_mutex> transLock(transportHeterogMap_.GetMtx());
        auto iter = transportHeterogMap_.FindLockFree(tag);
        if (iter != transportHeterogMap_.EndLockFree() &&
            (iter->second.find(node.rankId) != iter->second.end())) {
            return HCCL_SUCCESS;
        }

        auto itEmp = transportHeterogMap_.EmplaceLockFree(tag, map<u32, unique_ptr<TransportHeterog>>());
        // 先占位
        itEmp.first->second[node.rankId] = nullptr;
    }

    std::unique_ptr<TransportHeterog> transportPtr = nullptr;
    if (isServerInner) {
        ret = BuildEsInnerTransport(machinePara, hostPort, tag, node, transportPtr);
    } else {
        ret = BuildEsInterTransport(machinePara, nodePort, tag, node, transportPtr);
    }

    auto updateTrans = [this] (map<u32, unique_ptr<TransportHeterog>> &transMap, u32 rankId,
        unique_ptr<TransportHeterog> &transportPtr) -> void {
        transMap[rankId] = move(transportPtr);
    };
    transportHeterogMap_.EmplaceAndUpdate(tag, updateTrans, node.rankId, transportPtr);

    if (ret == HCCL_E_TIMEOUT) {
        ExceptionTransportInfoRetriever(realUserRank_, transportHeterogMap_[tag]);
    }

    return ret;
}

HcclResult HcclImplBase::BuildTransport(const u32 &nodePort, const u32 &hostPort, const s32 &tag,
    const std::vector<RoleTableNodeInfo> &nodesInfo)
{
    MachinePara machinePara;
    machinePara.localIpAddr = devIpAddr_[0];
    machinePara.localUserrank = userRank_;
    machinePara.localWorldRank = userRank_;
    machinePara.collectiveId = collectiveId_;
    machinePara.inputMem = csInitDefaultMem_;
    machinePara.outputMem = csInitDefaultMem_;
    machinePara.tag = std::to_string(tag);
    machinePara.deviceLogicId = deviceLogicId_;

    // 首先建链pcie
    for (const auto &node : nodesInfo) {
        HCCL_INFO("[BuildTransport]nodesInfo tag[%d] serverId_[%s] node.serverId[%s] node.rank[%u]", tag, serverId_.c_str(), node.serverId.c_str(), node.rankId);
        if (serverId_ == node.serverId) {
            CHK_RET(CreateAndSaveTransport(machinePara, nodePort, hostPort, tag, node, true));
        }
    }

    // 再建链roce
    for (const auto &node : nodesInfo) {
        HCCL_INFO("[BuildTransport]nodesInfo tag[%d] serverId_[%s] node.serverId[%s] node.rank[%u]", tag, serverId_.c_str(), node.serverId.c_str(), node.rankId);
        if (serverId_ != node.serverId) {
            CHK_RET(CreateAndSaveTransport(machinePara, nodePort, hostPort, tag, node, false));
        }
    }

    HCCL_INFO("BuildTransport tag[%d] Num[%d]", tag, transportHeterogMap_[tag].size());
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::SetQosCfg(const u32 qosCfg)
{
    CHK_SMART_PTR_NULL(dispatcher_);
    return HcclSetQosCfg(dispatcher_, qosCfg);
}

HcclResult HcclImplBase::ResetQosCfg()
{
    CHK_SMART_PTR_NULL(dispatcher_);
    return HcclResetQosCfg(dispatcher_);
}

HcclResult HcclImplBase::GetQosCfg(u32& qosCfg)
{
    CHK_SMART_PTR_NULL(dispatcher_);
    return HcclGetQosCfg(dispatcher_, &qosCfg);
}

HcclResult HcclImplBase::SetDeviceThread(bool isUseDevMap)
{
    rtContext_t ctx = nullptr;
    CHK_RET(hrtCtxGetCurrent(&ctx));
    HCCL_INFO("[SetDeviceThread]ctx[%p]", ctx);
    if (isUseDevMap) {
        const auto &devIpInfo = rankDevicePhyIdNicInfoMap_[hostIp_.GetReadableAddress()];
        for (const auto &rankInfo : devIpInfo) {
            CHK_RET(CreateCtxAndSetDev(rankInfo.first));
        }
    } else {
        for (const auto &rankInfo : rankInfoList_) {
            if (hostIp_ == rankInfo.hostIp && rankInfo.devicePhyId != HOST_DEVICE_ID) {
                CHK_RET(CreateCtxAndSetDev(rankInfo.devicePhyId));
            }
        }
    }
    if (ctx != nullptr) {
        CHK_RET(hrtCtxSetCurrent(ctx));
    }
    HCCL_INFO("[SetDeviceThread] exit");
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CreateCtxAndSetDev(u32 devicePhyId)
{
    rtContext_t tmpCtx = nullptr;
    u32 devLogicId;
    CHK_RET(hrtGetDeviceIndexByPhyId(devicePhyId, devLogicId));
    HCCL_INFO("SetDevice devPhyId[%d] devLogicId[%d].", devicePhyId, devLogicId);
    CHK_RET(hrtCtxCreate(&tmpCtx, RT_CTX_NORMAL_MODE, devLogicId));
    rtCtxMap_[devLogicId] = tmpCtx;
    CHK_RET(hrtSetDevice(devLogicId));
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::SetDevice(bool isUseDevMap)
{
    if (role_ == SERVER_ROLE_SOCKET) {
        HCCL_INFO("[SetDevice] entry");
        std::unique_ptr<std::thread> setDeviceThreadPtr;
        setDeviceThreadPtr.reset(new (std::nothrow) std::thread(&HcclImplBase::SetDeviceThread, this, isUseDevMap));

        CHK_PRT_RET(!setDeviceThreadPtr, HCCL_ERROR("[SetDevice]setDeviceThreadPtr threads reset failed "),
            HCCL_E_INTERNAL);

        if (setDeviceThreadPtr->joinable()) {
            setDeviceThreadPtr->join();  // 等待线程执行完毕
        }
    }
    HCCL_INFO("[SetDevice] exit");
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::resetDeviceThread()
{
    HCCL_INFO("[resetDeviceThread]");
    for (const auto &rankInfo : rankInfoList_) {
        if (hostIp_ == rankInfo.hostIp && rankInfo.devicePhyId != HOST_DEVICE_ID) {
            u32 devLogicId;
            CHK_RET(hrtGetDeviceIndexByPhyId(rankInfo.devicePhyId, devLogicId));
            HCCL_INFO("reset device phy id[%d] logic id[%u], ctx[%p]", rankInfo.devicePhyId, devLogicId,
                rtCtxMap_[devLogicId]);
            CHK_RET(hrtResetDevice(devLogicId));
            CHK_RET(hrtCtxDestroy(rtCtxMap_[devLogicId]));
            rtCtxMap_[devLogicId] = nullptr;
        }
    }
    HCCL_INFO("[resetDeviceThread] exit");
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::ResetDevice()
{
    if (role_ == SERVER_ROLE_SOCKET) {
        HCCL_INFO("[ResetDevice] entry");

        std::unique_ptr<std::thread> resetDeviceThreadPtr;

        resetDeviceThreadPtr.reset(new (std::nothrow) std::thread(&HcclImplBase::resetDeviceThread, this));

        CHK_PRT_RET(!resetDeviceThreadPtr, HCCL_ERROR("[ResetDevice]threads reset failed "),
            HCCL_E_INTERNAL);

        if (resetDeviceThreadPtr->joinable()) {
            resetDeviceThreadPtr->join();  // 等待线程执行完毕
        }
        HCCL_INFO("[ResetDevice] exit");
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetDeviceNicInfo(RankInfo& rankInfo)
{
    // hdc模式下整理ranktable信息通过map的形式访问device网卡
    if (rankDevicePhyIdNicInfoMap_.find(rankInfo.hostIp.GetReadableAddress()) != rankDevicePhyIdNicInfoMap_.end()) {
        return HCCL_SUCCESS;
    }
    for (auto &nodeInfo : rankInfoList_) {
        if (rankInfo.hostIp == nodeInfo.hostIp && nodeInfo.devicePhyId != HOST_DEVICE_ID) {
            rankDevicePhyIdNicInfoMap_[rankInfo.hostIp.GetReadableAddress()][nodeInfo.devicePhyId] = nodeInfo.nicIp[0];
            HCCL_INFO("GetDeviceNicInfo ps ip[%s] get available nic ip[%s], key[%s][%u]",
                rankInfo.nicIp[0].GetReadableAddress(), nodeInfo.nicIp[0].GetReadableAddress(),
                rankInfo.hostIp.GetReadableAddress(), nodeInfo.devicePhyId);
        }
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CsInit(const RoleTableInfo &roleTableInfo)
{
    csInitDefaultMem_ = DeviceMem::alloc(HCCL_BARRIER_DEFAULT_COUNT * sizeof(float));
    CHK_PTR_NULL(csInitDefaultMem_.ptr());

    serversInfo_ = roleTableInfo.servers;
    clientsInfo_ = roleTableInfo.clients;
    u32 roleTableSize = serversInfo_.size() + clientsInfo_.size();
    std::vector<int> psVec;

    // 判断当前是否为host侧且使用device网卡
    CHK_RET(IsHostUseDevNic(isHostUseDevNic_));

    for (auto &serverInfo : serversInfo_) {
        for (auto &rankInfo : rankInfoList_) {
            if (rankInfo.nicIp[0] == serverInfo.ipAddr && ranksPort_[rankInfo.userRank] == serverInfo.port) {
                roleTableSize--;
                serverInfo.rankId = rankInfo.userRank;
                serverInfo.serverId = rankInfo.serverId;
                serverInfo.hostIp = rankInfo.hostIp;
                serverInfo.hostPort = serverInfo.port;
                serverInfo.devicePhyId = rankInfo.devicePhyId;
                psVec.push_back(serverInfo.id);
                HCCL_INFO("Ps id[%u] rankTable ip[%s], hostIp[%s], port[%u], roleTable ip[%s], port[%u],"
                    "rankId[%u], userRank[%u], devicePhyId[%d].",
                    serverInfo.id, rankInfo.nicIp[0].GetReadableAddress(), serverInfo.hostIp.GetReadableAddress(),
                    ranksPort_[rankInfo.userRank], serverInfo.ipAddr.GetReadableAddress(), serverInfo.port,
                    serverInfo.rankId, rankInfo.userRank, serverInfo.devicePhyId);
            } else {
                continue;
            }
        }
    }

    std::sort(serversInfo_.begin(), serversInfo_.end(),
        [](const RoleTableNodeInfo &left, const RoleTableNodeInfo &right)-> bool {return left.id < right.id;});
    std::sort(psVec.begin(), psVec.end());

    for (auto &clientInfo : clientsInfo_) {
        for (auto &rankInfo : rankInfoList_) {
            if (rankInfo.nicIp[0] == clientInfo.ipAddr && ranksPort_[rankInfo.userRank] == clientInfo.port) {
                roleTableSize--;
                clientInfo.rankId = rankInfo.userRank;
                clientInfo.serverId = rankInfo.serverId;
                clientInfo.hostIp = rankInfo.hostIp;
                clientInfo.hostPort = HOST_PARA_BASE_PORT + rankInfo.devicePhyId;
                clientInfo.devicePhyId = rankInfo.devicePhyId;
                HCCL_INFO("clients rankTable ip[%s], prot[%u], roleTable ip[%s], hostIp[%s] port[%u], "
                          "rankId[%u], userRank[%u], devicePhyId[%d].",
                    rankInfo.nicIp[0].GetReadableAddress(), ranksPort_[rankInfo.userRank],
                    clientInfo.ipAddr.GetReadableAddress(), clientInfo.hostIp.GetReadableAddress(), clientInfo.port,
                    clientInfo.rankId, rankInfo.userRank, rankInfo.devicePhyId);
            } else {
                continue;
            }
        }
    }

    CHK_PRT_RET(roleTableSize != 0, HCCL_ERROR("RoleTable cannot exactly match ranktable."), HCCL_E_PARA);

    // 获取每个ps可用的device nic信息，支持ranktable和rolatable长度不同
    for (auto &rankInfo : rankInfoList_) {
        if (rankInfo.devicePhyId == HOST_DEVICE_ID) {
            CHK_RET(GetDeviceNicInfo(rankInfo));
        }
    }

    for (const auto &node : serversInfo_) {
        HCCL_INFO("serversInfo_ size[%u], rankId[%u], userRank_[%u].", serversInfo_.size(), node.rankId, userRank_);
        if (node.rankId == userRank_) {
            role_ = SERVER_ROLE_SOCKET;
            if (isHostUseDevNic_) {
                CHK_RET(SetDevice());
            }

            CHK_RET(InitPsRaResource());
            if (!GetExternalInputHcclIsTcpMode()) {
                CHK_RET(InitPsRdmaHandle());
            }
            break;
        }
    }

    for (const auto &clientInfo : clientsInfo_) {
        HCCL_INFO("clientsInfo_ size[%u], rankId[%u], userRank_[%u].",
            clientsInfo_.size(), clientInfo.rankId, userRank_);
        if (clientInfo.rankId == userRank_) {
            u32 partitionNum = 65537;
            u32 psNum = psVec.size();
            u32 psId[psNum];
            std::copy(psVec.begin(), psVec.end(), psId);
            HCCL_INFO("[Init][CsComm]hrtInitPartitionMap partitionNum[%u] psNum[%u].", partitionNum, psNum);
            CHK_RET(hrtInitPartitionMap(partitionNum, psNum, psId));

            role_ = CLIENT_ROLE_SOCKET;

            if (!Is310PDevice()) {
                CHK_RET(InitWorkerRaResource());
                if (IsEnableRoce()) {
                    CHK_RET(NetWorkRdmaSetQpDepth());
                }
            }

            if (isHostUseDevNic_) {
                // hdc模式下hccp默认开启白名单校验，worker将使用的device网卡添加至白名单中
                CHK_RET(InitHdcWhlistAdd());
            }
            break;
        }
    }

    CHK_RET(InitRecvMsgAndRequestBuffer());
    csCommInitFlag_ = true;

#ifndef CCL_KERNEL
    CHK_RET(InitESProfiler());
#endif

    // 设置event scheduler的event id, CsInit的event与异构通信的区分开，使用310P
    LocalIpcNotify notify;
    notify.SetEventIdAndTid(AUX_EVENT_ID, ORIGINAL_THREAD_ID);

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CsDeInit()
{
    if (!csCommInitFlag_) {
        return HCCL_SUCCESS;
    }

    CHK_RET(DeinitAndClearTransportMap());
    CHK_RET(ClearPsTagNodesInfo());

    if (hostRdmaInitFlag_) {
        CHK_RET(DeInitTransportMem());
        CHK_RET(MrManagerDeInit());
    }

    CHK_RET(DeinitWorkerRaResource());
    CHK_RET(DeinitPsRaResource());

    if (csInitDefaultMem_.ptr() != nullptr) {
        csInitDefaultMem_.free();
    }

    serversInfo_.clear();
    clientsInfo_.clear();
    csCommInitFlag_ = false;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::RegEsWorkSpaceResource(HcclOpPtr opPtr)
{
    CHK_RET(CreateOrUpdateTransferMem(opPtr));

    HcclOperator<EmbeddingServiceParam> *ptr = static_cast<HcclOperator<EmbeddingServiceParam> *>(opPtr);
    u64 keyMaxNum = ptr->param_.keyMaxNum;

    // 注册stag
    CHK_RET(notifyPool_->RegisterOp(ptr->param_.sTag));

    if (keyMaxNum > ES_PIPELINE_THRESHOLD) {
        s32 cubeSize = keyMaxNum / ES_PIPELINE_THRESHOLD + (keyMaxNum % ES_PIPELINE_THRESHOLD != 0);
        // 申请notify，注册流
        CHK_RET(RegSubStreamAndNotify(opPtr, cubeSize));
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::RegSubStreamAndNotify(HcclOpPtr opPtr, s32 cubeSize)
{
    HcclOperator<EmbeddingServiceParam> *ptr = static_cast<HcclOperator<EmbeddingServiceParam> *>(opPtr);

    std::string &sTag = ptr->param_.sTag;
    // 申请notify个数：cubeSize * 3 + 2
    // cubeSize * 3是4个流水kernel做同步需要的notify
    // 2是reset去重句柄和wait send key finish 做同步需要的notify
    u32 notifyCount = cubeSize * ES_PIPELINE_NOFITY_COUNT + 2;
    if (tagStreamInfo_.count(sTag) != 0) {
        level1StreamInfo_t &streamInfo = tagStreamInfo_[sTag];
        ptr->param_.embeddingParam.subStreamInfo.resize(HCCL_SUB_STREAM_ES_LOOKUP);
        ptr->param_.embeddingParam.notifyInfo.resize(notifyCount);
        for (u32 i = 0; i < HCCL_SUB_STREAM_ES_LOOKUP; ++i) {
            ptr->param_.embeddingParam.subStreamInfo[i] = streamInfo.ringStreams[i].ptr();
        }
        for (u32 i = 0; i < notifyCount; ++i) {
            ptr->param_.embeddingParam.notifyInfo[i] = streamInfo.ringSignal[i].get();
        }
        return HCCL_SUCCESS;
    }
    level1StreamInfo_t streamInfo;
    streamInfo.ringStreams.resize(HCCL_SUB_STREAM_ES_LOOKUP);
    ptr->param_.embeddingParam.subStreamInfo.resize(HCCL_SUB_STREAM_ES_LOOKUP);

    streamInfo.ringSignal.resize(notifyCount);
    ptr->param_.embeddingParam.notifyInfo.resize(notifyCount);
    streamInfo.ringStreams = workSpaceRes_->AllocSlaveStreams(sTag, HCCL_SUB_STREAM_ES_LOOKUP);
    CHK_PRT_RET((streamInfo.ringStreams.size() != streamInfo.ringNum - 1),
        HCCL_ERROR("[Reg][SubStreamAndNotify]get worksapce stream, tag[%s], " \
        "expect to get size [%d], but only get [%d].",
        sTag.c_str(), HCCL_SUB_STREAM_ES_LOOKUP, streamInfo.ringStreams.size()), HCCL_E_INTERNAL);
    for (u32 i = 0; i < HCCL_SUB_STREAM_ES_LOOKUP; ++i) {
        ptr->param_.embeddingParam.subStreamInfo[i] = streamInfo.ringStreams[i].ptr();
        if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
            CHK_RET(StreamActiveManager::GetInstance(deviceLogicId_).StreamActive(streamInfo.ringStreams[i].ptr(),
                ptr->param_.embeddingParam.stream));
        }
    }

    CHK_RET(queueNotifyManager_->Alloc(sTag, notifyCount, streamInfo.ringSignal));
    for (u32 i = 0; i < notifyCount; ++i) {
        ptr->param_.embeddingParam.notifyInfo[i] = streamInfo.ringSignal[i].get();
    }
    tagStreamInfo_.insert({sTag, streamInfo});

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CreateOrUpdateTransferMem(HcclOpPtr opPtr)
{
    HcclOperator<EmbeddingServiceParam> *ptr = static_cast<HcclOperator<EmbeddingServiceParam> *>(opPtr);
    lock_guard<std::mutex> lockTransferMemsMap(transferMemsMapMutex_);
    auto tag = ptr->param_.tag;

    if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE && transferMemsMap_.count(tag) != 0) {
        HCCL_INFO("[Create][TransferMem]tag is existed, tag(%u), GetWorkflowMode=%u", tag,
            static_cast<u32>(GetWorkflowMode()));
        return HCCL_SUCCESS;
    }

    transferMemsMap_[tag].resize(TRANSFER_MEM_INFO_IDX_NUM);
    // +1是为table id申请的内存
    u64 keyMemSize = (((ptr->param_.keyMaxNum + 1) * SIZE_TABLE[ptr->param_.keyType] / IPC_MEM_ALIGNMENT_BYTE) + 1) *
        IPC_MEM_ALIGNMENT_BYTE;
    // 为key的中转申请2倍内存,1份用作shard多线程去重，1份用作send key buf
    keyMemSize += keyMemSize;

    if (ptr->param_.embeddingParam.flags == ES_FLAGS_ENABLE_COUNTER) {
        u64 counterSize = Align<u64>(static_cast<u64>(ptr->param_.keyMaxNum) * ES_KEY_COUNTER_MEM_BYTES_SIZE,
            IPC_MEM_ALIGNMENT_BYTE);
        keyMemSize += counterSize;
    }
    s32 maxValueItemSize = ptr->param_.maxEmbeddingDim * SIZE_TABLE[ptr->param_.valuesType];
    u64 valueMemSize = ((ptr->param_.keyMaxNum * maxValueItemSize /
        IPC_MEM_ALIGNMENT_BYTE) + 1) * IPC_MEM_ALIGNMENT_BYTE;
    u64 rdmaEnvelopeMemSize = Align<u64>(sizeof(HcclEsRdmaInfo) * ES_MAX_PS_NUM, IPC_MEM_ALIGNMENT_BYTE);

    vector<DeviceMem> &transferMem = transferMemsMap_[tag];
    if (ptr->param_.uniqueTag.empty()) {
        ptr->param_.uniqueTag = std::to_string(tag);
    }
    if (esWorkMems_.find(ptr->param_.uniqueTag) != esWorkMems_.end() &&
        GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
        transferMem[TRANSFER_MEM_INFO_KEY_IDX] =
            DeviceMem::create(esWorkMems_[ptr->param_.uniqueTag], keyMemSize);
        transferMem[TRANSFER_MEM_INFO_VALUE_IDX] =
            DeviceMem::create(reinterpret_cast<void *>(static_cast<u8 *>(esWorkMems_[ptr->param_.uniqueTag]) +
            keyMemSize), valueMemSize);
        transferMem[TRANSFER_MEM_INFO_RDMA_ENVELOPE_IDX] =
            DeviceMem::create(reinterpret_cast<void *>(static_cast<u8 *>(esWorkMems_[ptr->param_.uniqueTag]) +
            keyMemSize + valueMemSize), rdmaEnvelopeMemSize);
    } else {
        transferMem[TRANSFER_MEM_INFO_KEY_IDX] =
            workSpaceRes_->AllocDeviceMem(std::to_string(tag), keyMemSize);
        transferMem[TRANSFER_MEM_INFO_VALUE_IDX] =
            workSpaceRes_->AllocDeviceMem(std::to_string(tag), valueMemSize);
        transferMem[TRANSFER_MEM_INFO_RDMA_ENVELOPE_IDX] =
            workSpaceRes_->AllocDeviceMem(std::to_string(tag), rdmaEnvelopeMemSize);
        esWorkMems_[ptr->param_.uniqueTag] = transferMem[TRANSFER_MEM_INFO_KEY_IDX].ptr();
    }
    CHK_PTR_NULL(transferMem[TRANSFER_MEM_INFO_KEY_IDX].ptr());
    CHK_PTR_NULL(transferMem[TRANSFER_MEM_INFO_VALUE_IDX].ptr());
    CHK_PTR_NULL(transferMem[TRANSFER_MEM_INFO_RDMA_ENVELOPE_IDX].ptr());

    HCCL_INFO("tag[%d], keyMemAddr[%llu], keyMemSize[%llu], valueMemAddr[%llu], valueMemSize[%llu], envelopeMemAddr"
        "[%llu], rdmaEnvelopeMemSize[%llu]", tag, hash<void *>{}(transferMem[TRANSFER_MEM_INFO_KEY_IDX].ptr()),
        keyMemSize, hash<void *>{}(transferMem[TRANSFER_MEM_INFO_VALUE_IDX].ptr()), valueMemSize,
        hash<void *>{}(transferMem[TRANSFER_MEM_INFO_RDMA_ENVELOPE_IDX].ptr()), rdmaEnvelopeMemSize);

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetPsMap(std::map<u32, u32> &nodeMap)
{
    for (auto &serverInfo : serversInfo_) {
        nodeMap[serverInfo.id] = serverInfo.rankId;
        HCCL_INFO("ps rankId[%u], psId[%u].", serverInfo.rankId, serverInfo.id);
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetTagWorkerList(u32 tag, std::vector<u32> &nodeList)
{
    if (isEsAllLinkModel_) {
        HCCL_WARNING("isEsAllLinkModel_ is true, need get all ranklist");
        CHK_RET(GetWorkerList(nodeList));
        return HCCL_SUCCESS;
    }

    nodeList.clear();
    std::unique_lock<std::mutex> tagNodesInfoLock(tagNodesInfoLock_);
    if (tagNodesInfo_.find(tag) == tagNodesInfo_.end()) {
        HCCL_ERROR("tagNodesInfo_ can't find tag[%u]", tag);
        return HCCL_E_PARA;
    }

    for (auto &nodeInfo : tagNodesInfo_[tag]) {
        nodeList.push_back(nodeInfo.rankId);
        HCCL_INFO("tag[%u] worker rankId[%u].", tag, nodeInfo.rankId);
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetWorkerList(std::vector<u32> &nodeList)
{
    nodeList.clear();
    for (auto &clientInfo : clientsInfo_) {
        nodeList.push_back(clientInfo.rankId);
        HCCL_INFO("worker rankId[%u].", clientInfo.rankId);
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::AsyncCopyHostEnvelopeToDev(const HdcsEmbeddingServiceParam &para, void *stream)
{
    if (!para.haveRdmaConn) {
        HCCL_INFO("haveRdmaConn is false, no need to AsyncCopyHostEnvelopeToDev");
        return HCCL_SUCCESS;
    }

    HcclEsRdmaInfoForUpdate *devRdmaEnveInfos = static_cast<HcclEsRdmaInfoForUpdate *>(para.rdmaEnveInfosTransferMem);
    unique_ptr<HcclEsRdmaInfoForUpdate[]> &rdmaEnveInfos = rdmaEnvelopeInfoHostMems_[para.tag];

    u64 maxMemSize = sizeof(HcclEsRdmaInfoForUpdate) * ES_MAX_PS_NUM;
    if (devRdmaEnveInfos == nullptr || para.rdmaEnveInfosTransferMemSize < maxMemSize) {
        HCCL_ERROR("rdmaEnveInfosTransferMem is nullptr, or rdmaEnveInfosTransferMemSize[%llu] < maxMemSize[%llu]",
            para.rdmaEnveInfosTransferMemSize, maxMemSize);
        return HCCL_E_PARA;
    }

    u64 memSize = sizeof(HcclEsRdmaInfoForUpdate) * serversInfo_.size();
    CHK_RET(hrtMemAsyncCopy(devRdmaEnveInfos, maxMemSize, rdmaEnveInfos.get(),
        memSize, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE, stream));

    PRINT_ARRAY(reinterpret_cast<u64 *>(rdmaEnveInfos.get()), memSize / sizeof(u64), "UpdateHostRdmaEnveInfos");

    HCCL_DEBUG("AsyncCopyHostEnvelopeToDev SUCCESS !");
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::FillEnvelope(const HdcsEmbeddingServiceParam &para, u32 rankId, HcclEnvelope &envelope)
{
    TransportHeterogRoce *transportRoce =
        dynamic_cast<TransportHeterogRoce *>(transportHeterogMap_[para.tag][rankId].get());
    CHK_PTR_NULL(transportRoce);
    void *workerspaceAddr = para.keyTransferMem;

    u32 lkey = 0;
    CHK_RET(transportRoce->RegMr(workerspaceAddr, para.workerspaceMemSize, lkey, true));
    CHK_RET(transportRoce->RegMr(workerspaceAddr, para.workerspaceMemSize, lkey, false));

    HCCL_DEBUG("RegMr workerspace mem size[%llu], rdmaEnveInfosTransferMemSize[%llu]",
        para.workerspaceMemSize, para.rdmaEnveInfosTransferMemSize);

    TransportEndPointInfo srcEp(0, 0, para.tag);
    TransportEndPointInfo dstEp(0, rankId, para.tag);

    // envelope.transData 由device aicpu填写
    envelope.epParam.src = srcEp;
    envelope.epParam.dst = dstEp;
    envelope.protocol = 0;
    envelope.key = lkey;
    envelope.msn = 0;

    RdmaBuffer *rdmaBuffer = reinterpret_cast<RdmaBuffer *>(envelope.rsv);

    // addr内容 由device aicpu填写回写地址
    rdmaBuffer->addr = 0;
    // 填写回写地址的key
    rdmaBuffer->buffKey = lkey;

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::PrepareHostEnvelopeTemplates(HdcsEmbeddingServiceParam &para)
{
    if (rdmaEnvelopeInfoHostMems_[para.tag] == nullptr) {
        EXECEPTION_CATCH((rdmaEnvelopeInfoHostMems_[para.tag] = make_unique<HcclEsRdmaInfoForUpdate[]>(ES_MAX_PS_NUM)),
            return HCCL_E_PTR);
    }

    unique_ptr<HcclEsRdmaInfoForUpdate[]> &rdmaEnveInfos = rdmaEnvelopeInfoHostMems_[para.tag];
    para.haveRdmaConn = false;
    for (u32 i = 0; i < serversInfo_.size(); ++i) {
        u32 rankId = serversInfo_[i].rankId;
        u32 psId = i;

        TransportHeterog *transportHeterog = transportHeterogMap_[para.tag][rankId].get();
        CHK_PTR_NULL(transportHeterog);
        TransportShmEvent *transport = dynamic_cast<TransportShmEvent *>(transportHeterog);
        if (transport == nullptr) {
            TransportHeterogRoce *transportRoce = dynamic_cast<TransportHeterogRoce *>(transportHeterog);
            CHK_PRT_RET(transportRoce == nullptr,
                HCCL_ERROR("rankId[%u] transport type is not TransportShmEvent or TransportHeterogRoce.",
                rankId), HCCL_E_PARA);

            para.haveRdmaConn = true;
            rdmaEnveInfos[psId].commType = HcclHeterogCommType::RDMA;

            CHK_RET(FillEnvelope(para, rankId, rdmaEnveInfos[psId].envelope));

            // 为update的value信封模板进行拷贝
            rdmaEnveInfos[psId].envelopeValue = rdmaEnveInfos[psId].envelope;

            HCCL_INFO("psId[%u], rank[%u] transport type is TransportHeterogRoce", psId, rankId);
        } else {
            rdmaEnveInfos[psId].commType = HcclHeterogCommType::PCIE;
            HCCL_INFO("psId[%u], rank[%u] transport type is TransportShmEvent", psId, rankId);
        }
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::RegEsPsTransferMem(const HcclOperator<EmbeddingServiceParam> &op)
{
    AlgCollRemoteUpdate *algUpdate = nullptr;

    EXECEPTION_CATCH({ auto &alg = GenerateEmbAlg<AlgCollRemoteUpdate>(op.param_.tag); \
        algUpdate = dynamic_cast<AlgCollRemoteUpdate *>(alg.get()); \
        }, return HCCL_E_PTR);

    CHK_SMART_PTR_NULL(algUpdate);

    CHK_RET(algUpdate->AllocTransferMemAndRegMrForPs(op));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::MrManagerInit()
{
    // 拉远、下沉、推理场景(ps、worker 通用服务器)支持使用mrManager
    if (!GetExternalInputHcclIsTcpMode() && (isHeterogComm_ || Is310PDevice() || IsGeneralServer())) {
        // 构造中会进行mrManager的reset，且将其保存到transport层，此处为了避免智能指针和裸指针混用，增加判断避免重复reset
        if (!IsGeneralServer()) {
            mrManager_.reset(new (std::nothrow) MrManager());
            CHK_SMART_PTR_NULL(mrManager_);
        } else {
            if (mrManager_ == nullptr) {
                mrManager_.reset(new (std::nothrow) MrManager());
                CHK_SMART_PTR_NULL(mrManager_);
            }
        }

        bool isLocalDevNicInPlane;
        CHK_RET(IsLocalDevNicInPlane(isLocalDevNicInPlane));
        RaResourceInfo raResourceInfo{};
        // 如果本端不分平面而且为AiServer场景的ps的话，从此进入
        if (!isLocalDevNicInPlane && isHostUseDevNic_ && deviceLogicId_ == HOST_DEVICE_ID) {
            u32 devLogicId;
            HcclIpAddress nicIp;

            u32 devicePhyId;
            CHK_PRT(GetDevIP(userRank_, hostIp_, devicePhyId, nicIp));
            CHK_RET(hrtGetDeviceIndexByPhyId(devicePhyId, devLogicId));

            CHK_RET(NetworkManager::GetInstance(devLogicId).GetRaResourceInfo(raResourceInfo));
            void *nicRdmaHandle = raResourceInfo.nicSocketMap[nicIp].nicRdmaHandle;

            CHK_RET(mrManager_->Init(nicRdmaHandle, devLogicId, true));
        } else {
            CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
            void *nicRdmaHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicRdmaHandle;
            CHK_RET(mrManager_->Init(nicRdmaHandle));
        }

        mrManagerInit_ = true;
    }
    transportResourceInfo_.isGlobalMrmanagerInit = mrManagerInit_;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::MrManagerDeInit()
{
    if (mrManagerInit_) {
        RaResourceInfo raResourceInfo;
        u32 devLogicId = deviceLogicId_;
        HcclIpAddress nicIp = devIpAddr_[0];
        if (isHostUseDevNic_ && deviceLogicId_ == HOST_DEVICE_ID) {
            u32 devicePhyId;
            CHK_PRT(GetDevIP(userRank_, hostIp_, devicePhyId, nicIp));
            CHK_RET(hrtGetDeviceIndexByPhyId(devicePhyId, devLogicId));
        }
        CHK_RET(NetworkManager::GetInstance(devLogicId).GetRaResourceInfo(raResourceInfo));
        void *nicRdmaHandle = raResourceInfo.nicSocketMap[nicIp].nicRdmaHandle;

        CHK_SMART_PTR_NULL(mrManager_);
        CHK_RET(mrManager_->DeInit(nicRdmaHandle));
        mrManager_ = nullptr;
        mrManagerInit_ = false;
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::SupportDeterministicOptim(bool &isDeterministicOptim)
{
    CHK_SMART_PTR_NULL(implAlg_);
    CHK_RET(implAlg_->SupportDeterministicOptim(isDeterministicOptim));
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetHccsLinkNum(u32 &numHccsLink)
{
    auto iter = pairLinkInfo_.find(static_cast<u32>(LinkTypeInServer::HCCS_TYPE));
    if (iter == pairLinkInfo_.end()) {
        HCCL_ERROR("[HcclImplBase][GetHccsLinkNum]HCCS_TYPE is not found");
        return HCCL_E_PARA;
    }
    numHccsLink = iter->second.size();
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::ServiceCancel(HcclOpPtr opPtr)
{
    HcclOperator<EmbeddingServiceParam> *ptr = static_cast<HcclOperator<EmbeddingServiceParam> *>(opPtr);

    auto memIt = tagRankTransMemsMap_.Find(ptr->param_.tag);
    if (memIt.second) {
        ptr->param_.getRequestParam.transferMems = memIt.first->second;
    }

    auto it = tagAlgMap_.Find(ptr->param_.tag);
    if (it.second) {
        auto &alg = it.first->second;
        CHK_SMART_PTR_NULL(alg);
        CHK_RET(alg->Process(opPtr));
        CHK_RET(tagOpExecutorMap_[ptr->param_.tag]->ExecuteOp(opPtr));
    } else {
        AlgBase *algPtr = static_cast<AlgBase *>(ptr->param_.algPtr);
        CHK_PTR_NULL(algPtr);
        CHK_RET(algPtr->Process(opPtr));
        CHK_RET(static_cast<OpExecutor *>(ptr->param_.opExecutorPtr)->ExecuteOp(opPtr));
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::RegisterShareMemory()
{
    s32 curPid = SalGetPid();
    GroupQueryOutput groupInfo = {0};
    u32 groupInfoLen = 0;

    CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
    CHK_RET(hrtHalGrpQuery(GRP_QUERY_GROUPS_OF_PROCESS, &curPid, static_cast<u32>(sizeof(curPid)),
        reinterpret_cast<void *>(&groupInfo), &groupInfoLen));

    if ((groupInfoLen == 0) || (groupInfoLen % sizeof(groupInfo.grpQueryGroupsOfProcInfo[0]) != 0)) {
        HCCL_INFO("[RegisterShareMemory] group info size[%u] is invalid", groupInfoLen);
        return HCCL_SUCCESS;
    }
    u32 groupNum = static_cast<u32>(groupInfoLen / sizeof(groupInfo.grpQueryGroupsOfProcInfo[0]));
    if (groupNum >= BUFF_GRP_MAX_NUM) {
        HCCL_ERROR("[RegisterShareMemory] groupNum [%u] is bigger than [%u]", groupNum, BUFF_GRP_MAX_NUM);
        return HCCL_E_PARA;
    }
    for (u32 i = 0; i < groupNum; i++) {
        GrpQueryGroupAddrPara queryPara = {};
        CHK_SAFETY_FUNC_RET(memcpy_s(queryPara.grpName, BUFF_GRP_NAME_LEN,
            groupInfo.grpQueryGroupsOfProcInfo[i].groupName, BUFF_GRP_NAME_LEN));
        queryPara.devId = static_cast<u32>(deviceLogicId_);
        GrpQueryGroupAddrInfo queryResults[BUFF_GROUP_ADDR_MAX_NUM] = {};
        u32 resultSize = 0;
        CHK_RET(hrtHalGrpQuery(GRP_QUERY_GROUP_ADDR_INFO, &queryPara, static_cast<u32>(sizeof(queryPara)),
            queryResults, &resultSize));
        u32 resultLen = static_cast<u32>(resultSize / sizeof(GrpQueryGroupAddrInfo));
        if (resultLen >= BUFF_GROUP_ADDR_MAX_NUM) {
            HCCL_ERROR("[RegisterShareMemory] resultLen [%u] is bigger than [%u]", resultLen, BUFF_GROUP_ADDR_MAX_NUM);
            return HCCL_E_PARA;
        }
        for (u32 index = 0; index < resultLen; index++) {
            HCCL_DEBUG("[RegisterShareMemory]addr[%p], size[%llu]", queryResults[index].addr, queryResults[index].size);
            if (!GetExternalInputHcclIsTcpMode()) {
                CHK_RET(mrManager_->RegGlobalMr(reinterpret_cast<void *>(queryResults[index].addr),
                    queryResults[index].size));
            }
        }
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::SetMeshAggregationRankSize(u32 size)
{
    HCCL_INFO("[Set][HcclImplBase][MeshAggregationRankSize]set MeshAggregationRankSize[%u].", size);
    meshAggregationRankSize_ = size;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitRecvMsgAndRequestBuffer()
{
    // 拉远、下沉、推理场景(ps、worker)支持使用msg/request内存池
    if (pMsgInfosMem_ == nullptr) {
        pMsgInfosMem_.reset(new (std::nothrow) LocklessRingMemoryAllocate<HcclMessageInfo>(MEMORY_CAPACITY));
        CHK_SMART_PTR_NULL(pMsgInfosMem_);
        CHK_RET(pMsgInfosMem_->Init());
        HCCL_INFO("InitRecvMsgBuffer Success!");
    }

    if (pReqInfosMem_ == nullptr) {
        pReqInfosMem_.reset(new (std::nothrow) LocklessRingMemoryAllocate<HcclRequestInfo>(MEMORY_CAPACITY));
        CHK_SMART_PTR_NULL(pReqInfosMem_);
        CHK_RET(pReqInfosMem_->Init());
        HCCL_INFO("InitRequestBuffer Success!");
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DeInitTransportMem()
{
    if (memBlocksManager_ != nullptr) {
        // 解注册内存
        CHK_RET(mrManager_->ReleaseKey(memBlocksManager_->GetMemAddr(), memBlocksManager_->GetMemSize()));
        memBlocksManager_ = nullptr;
    }

    if (pMsgInfosMem_ != nullptr) {
        pMsgInfosMem_ = nullptr;
    }

    if (pReqInfosMem_ != nullptr) {
        pReqInfosMem_ = nullptr;
    }

    if (pRecvWrInfosMem_ != nullptr) {
        pRecvWrInfosMem_ = nullptr;
    }
    HCCL_INFO("DeInitTransportMem Success!");
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitMemBlocksAndRecvWrMem()
{
    u32 memBlockNum = MEM_BLOCK_NUM;
    CHK_PRT(GetMemBlockNum(devicePhyId_, memBlockNum));

    if (!GetExternalInputHcclIsTcpMode() &&
        (isHeterogComm_ || Is310PDevice() || isHostUseDevNic_ || IsGeneralServer())) {
        // 注册mr,hdc模式下在通信类内进行
        if (IsNeedGlobalMrManager()) {
            // 初始化信封内存
            memBlocksManager_.reset(new (std::nothrow) HeterogMemBlocksManager());
            CHK_SMART_PTR_NULL(memBlocksManager_);
            CHK_RET(memBlocksManager_->Init(memBlockNum));

            // 信封内存注册
            CHK_RET(mrManager_->GetKey(memBlocksManager_->GetMemAddr(), memBlocksManager_->GetMemSize(),
                transportResourceInfo_.lkey));
        }

        // 初始化wr内存
        if (pRecvWrInfosMem_.get() == nullptr) {
            pRecvWrInfosMem_.reset(new (std::nothrow) LocklessRingMemoryAllocate<RecvWrInfo>(MEMORY_CAPACITY));
            CHK_SMART_PTR_NULL(pRecvWrInfosMem_);
            CHK_RET(pRecvWrInfosMem_->Init());
            HCCL_INFO("InitMemBlocksAndRecvWrMem Success!");
        }
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::CreateSrq()
{
    u32 info = 0;
    CHK_RET(hrtDrvGetPlatformInfo(&info));
    if (info == 0) {
        std::string chipName;
        HcclResult ret = hrtHalGetChipInfo(devicePhyId_, chipName);
        if (ret == HCCL_SUCCESS) {
            if (chipName.find(SOC_NAME_910B) != std::string::npos) {
                HCCL_INFO("not support chip[%s] create srq", chipName.c_str());
                return HCCL_SUCCESS;
            }
        }
    }

    if (!srqInit_ && !GetExternalInputHcclIsTcpMode() && (isHeterogComm_ || Is310PDevice())) {
        RaResourceInfo raResourceInfo;
        CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
        void *nicRdmaHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicRdmaHandle;

        // 创建srq
        transportResourceInfo_.tagSrqInfo.srqEvent = HCCL_EVENT_RECV_REQUEST_MSG;
        transportResourceInfo_.dataSrqInfo.srqEvent = HCCL_EVENT_SEND_COMPLETION_MSG;

        transportResourceInfo_.tagSrqInfo.srqDepth = MAX_SRQ_DEPTH;
        transportResourceInfo_.dataSrqInfo.srqDepth = MAX_SRQ_DEPTH;
        CHK_RET(hrtRaCreateSrq(nicRdmaHandle, transportResourceInfo_.tagSrqInfo));
        CHK_RET(hrtRaCreateSrq(nicRdmaHandle, transportResourceInfo_.dataSrqInfo));
        HCCL_INFO("CreateSrq Success!");

        std::unique_ptr<TransportHeterogEventRoce> transportPtr;
        transportPtr.reset(new (std::nothrow) TransportHeterogEventRoce(transportResourceInfo_));
        CHK_SMART_PTR_NULL(transportPtr);
        CHK_RET(transportPtr->InitSrqRecvWqe());

        srqInit_ = true;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::DestroySrq()
{
    if (srqInit_) {
        RaResourceInfo raResourceInfo;
        CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
        void *nicRdmaHandle = raResourceInfo.nicSocketMap[devIpAddr_[0]].nicRdmaHandle;

        // 销毁srq
        CHK_RET(hrtRaDestroySrq(nicRdmaHandle, transportResourceInfo_.tagSrqInfo));
        CHK_RET(hrtRaDestroySrq(nicRdmaHandle, transportResourceInfo_.dataSrqInfo));
        transportResourceInfo_.tagSrqInfo = SrqInfo();
        transportResourceInfo_.dataSrqInfo = SrqInfo();
        HCCL_INFO("DestroySrq Success!");
        srqInit_ = false;
    }

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::SetDevicePid(s32 devicePid)
{
    devicePid_ = devicePid;
    return HCCL_SUCCESS;
}

void HcclImplBase::ReleaseWorkSpacebuffer()
{
    workSpace_.free();
}

HcclResult HcclImplBase::CreateWorkSpace(u64 size, DeviceMem &buffer) const
{
    CHK_PRT_RET(!size, HCCL_INFO("[Create][WorkSpace]work space size is zero. not need to malloc memory"),
        HCCL_SUCCESS);

    CHK_PRT_RET((size > ULONG_MAX), \
        HCCL_ERROR("[Create][WorkSpace]work space size is greater than %llu",
            ULONG_MAX), HCCL_E_PARA);

    u64 memSize = size;
    buffer = DeviceMem::alloc(memSize);
    CHK_PRT_RET(size && !buffer, HCCL_ERROR("[Create][WorkSpace]Create work space size[%llu] fail,"\
        "please check workspace size.", size), HCCL_E_PTR);
    CHK_RET(hrtMemSet(buffer.ptr(), size, size));
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetWorkSpace(u64 *workSpaceSize, u64 *workSpace) const
{
    *workSpaceSize = workSpaceSize_;
    *workSpace = reinterpret_cast<u64>(workSpace_.ptr());
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::InitWorkSpace()
{
    if (workSpace_.ptr() == nullptr) {
        workSpaceSize_ = MAX_WORK_SPACE_SIZE;
        CHK_RET(CreateWorkSpace(workSpaceSize_, workSpace_));
    }

    return HCCL_SUCCESS;
}

void HcclImplBase::PrepareCommAsync()
{
    HcclResult ret;
    HCCL_RUN_INFO("thread PrepareCommAsync start.");
    do {
        if (threadPrepareCommAsyncExit_) {
            return;
        }
        u32 ongoingNum = 0;
        std::unique_lock<std::mutex> requestLock(prepareRequestMutex_);
        for (auto &iter : prepareRequest_) {
            if (iter.second.status == 0xFFFFFFFF) {
                ret = PrepareCommStart(iter.first, iter.second.inputPtr, iter.second.count, iter.second.dataType,
                    iter.second.peerRank, iter.second.status);
                if (ret != HCCL_SUCCESS) {
                    iter.second.status = HETEROG_P2P_FAILED;
                    HCCL_ERROR("PrepareCommStart failed. tag:%s", iter.first.c_str());
                    return;
                }
                HCCL_DEBUG("PrepareCommStart tag:%s", iter.first.c_str());
            } else if (iter.second.status == HETEROG_P2P_WAIT) {
                ret = PrepareCommQuerry(iter.first, iter.second.status);
                if (ret != HCCL_SUCCESS) {
                    iter.second.status = HETEROG_P2P_FAILED;
                    HCCL_ERROR("PrepareCommStart failed. tag:%s", iter.first.c_str());
                    return;
                }
                HCCL_DEBUG("PrepareCommQuerry tag:%s status:%u", iter.first.c_str(), iter.second.status);
            }
            if (iter.second.status == HETEROG_P2P_WAIT) {
                ongoingNum++;
            }
        }
        requestLock.unlock();
        if (ongoingNum == 0) {
            WaitPrepareComm();
        } else {
            SaluSleep(WAIT_PREPARE_SLEEP_TIME);
        }
    } while (1);
    return;
}

HcclResult HcclImplBase::PrepareCommStart(const std::string &tag, void *addr, u64 count, HcclDataType dataType,
    u32 peerRank, u32& status)
{
    if (!IsAtomicInit()) {
        HCCL_ERROR("[HcclImplBase][Send]errNo[0x%016llx] hccl initialization must be called before call this function",
            HCCL_ERROR_CODE(HCCL_E_UNAVAIL));
        return HCCL_E_UNAVAIL;
    }
    /* ------------集合通信资源准备------------ */
    u32 perDataSize = 0;
    CHK_RET(SalGetDataTypeSize(dataType, perDataSize));

    DeviceMem mem = DeviceMem::create(addr, static_cast<u64>(count) * perDataSize);

    // 创建通信域
    HCCL_INFO("resource creation start,tag[%s]", tag.c_str());
    CHK_RET(implAlg_->CreateP2PCommAsync(tag, mem, peerRank, status));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::PrepareCommQuerry(const std::string &tag, u32& status)
{
    CHK_SMART_PTR_NULL(implAlg_);
    CHK_RET(implAlg_->CreateP2PCommQuerry(tag, status));
    return HCCL_SUCCESS;
}

void HcclImplBase::NotifyPrepareComm()
{
    std::unique_lock<std::mutex> lock(startPrepareCommMtx_);
    startPrepareComm_.notify_one();
    return;
}

void HcclImplBase::WaitPrepareComm()
{
    std::unique_lock<std::mutex> lock(startPrepareCommMtx_);
    startPrepareComm_.wait(lock);
    return;
}

void HcclImplBase::StopPrepareComm()
{
    threadPrepareCommAsyncExit_ = true;
    NotifyPrepareComm();
    if (prepareCommThread_ != nullptr && prepareCommThread_->joinable()) {
        prepareCommThread_->join();
        prepareCommThread_ = nullptr;
    }
    return;
}

HcclResult HcclImplBase::PrepareGather(const std::string &tag, const HcomCollOpInfo &opInfo, HcomOperationType opType)
{
    CHK_SMART_PTR_NULL(implAlg_);
    u32 perDataSize = 0;
    CHK_RET(SalGetDataTypeSize(opInfo.dataType, perDataSize));

    DeviceMem inputMem = DeviceMem::create(opInfo.inputAddr, opInfo.count * perDataSize);
    DeviceMem outputMem;

    if (opInfo.root != userRank_) {
        if (!isServerInter_) {
            u64 workSpaceMemSize = opInfo.count * perDataSize * userRankSize_;
            if (tagWorkSpaceMem_.find(tag) == tagWorkSpaceMem_.end()) {
                tagWorkSpaceMem_[tag] = HostMem::alloc(workSpaceMemSize, false);
                CHK_SMART_PTR_NULL(tagWorkSpaceMem_[tag].ptr());
            }
            void *workSpaceMem = tagWorkSpaceMem_[tag].ptr();
            CHK_PTR_NULL(workSpaceMem);
            outputMem = DeviceMem::create(workSpaceMem, workSpaceMemSize);
        } else {
            outputMem = DeviceMem::create(opInfo.outputAddr, opInfo.count * perDataSize);
        }
    } else {
        outputMem = DeviceMem::create(opInfo.outputAddr, userRankSize_ * opInfo.count * perDataSize);
    }
    HCCL_DEBUG("inputMem[%p],outputMem[%p].", inputMem.ptr(), outputMem.ptr());
    if (isHaveCpuRank_ && !isSetHDCModeInfo_ && isServerInter_) {
        if (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
            CHK_RET(InitPsHdcRaResource());
            CHK_RET(InitHdcPsRdmaHandle());
        }
        isSetHDCModeInfo_ = true;
    }
    implAlg_->SetHDCModeInfo(rankDevicePhyIdNicInfoMap_, groupRanksPort_, isSetHDCModeInfo_, isUseRankPort_);

    // 创建通信域
    AlgType algType;
    algType.algoLevel0 = AlgTypeLevel0::ALG_LEVEL0_NP_STAR;
    algType.algoLevel1 = AlgTypeLevel1::ALG_LEVEL1_STAR;
    CHK_RET(implAlg_->SetAlgType(algType, HcclCMDType::HCCL_CMD_GATHER));
    HcclResult ret = implAlg_->CreateComm(tag, inputMem, outputMem, algType, opInfo.root);
    CHK_PRT_RET(ret != HCCL_SUCCESS,
        HCCL_ERROR("[HcclImplBase][AllGather]errNo[0x%016llx] tag[%s],all gather create comm failed",
        HCCL_ERROR_CODE(ret), tag.c_str()), ret);

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::PrepareBroadcast(
    const std::string &tag, const HcomCollOpInfo &opInfo, HcomOperationType opType)
{
    CHK_SMART_PTR_NULL(implAlg_);
    u32 perDataSize = 0;
    CHK_RET(SalGetDataTypeSize(opInfo.dataType, perDataSize));
    DeviceMem devMem = DeviceMem::create(opInfo.inputAddr, opInfo.count * perDataSize);

    if (isHaveCpuRank_ && !isSetHDCModeInfo_ && isServerInter_) {
        if (static_cast<s32>(devicePhyId_) == HOST_DEVICE_ID) {
            CHK_RET(InitPsHdcRaResource());
            CHK_RET(InitHdcPsRdmaHandle());
        }
        isSetHDCModeInfo_ = true;
    }
    implAlg_->SetHDCModeInfo(rankDevicePhyIdNicInfoMap_, groupRanksPort_, isSetHDCModeInfo_, isUseRankPort_);
    AlgType algType;
    algType.algoLevel0 = AlgTypeLevel0::ALG_LEVEL0_NP_STAR;
    algType.algoLevel1 = AlgTypeLevel1::ALG_LEVEL1_STAR;
    CHK_RET(implAlg_->SetAlgType(algType, HcclCMDType::HCCL_CMD_BROADCAST));
    CHK_RET(implAlg_->CreateComm(tag, devMem, devMem, algType, opInfo.root));

    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::SaveTraceInfo(std::string &logInfo)
{
    if (!isHeterogComm_) {
        opBaseAtraceInfo_->SaveTraceInfo(logInfo, AtraceOption::Opbasekey);
    } else {
        HCCL_RUN_INFO("%s", logInfo.c_str());
    }

    return HCCL_SUCCESS;
}

void HcclImplBase::Break()
{
    if (implAlg_ != nullptr) {
        implAlg_->Break();
    }
    return;
}

std::unordered_map<std::string, std::map<u32, HcclIpAddress>>& HcclImplBase::GetPhyIdNicInfo()
{
    return rankDevicePhyIdNicInfoMap_;
}

vector<u32> HcclImplBase::GetRanksPort()
{
    return ranksPort_;
}

vector<RankInfo> HcclImplBase::GetRanksList()
{
    return rankInfoList_;
}

HcclResult HcclImplBase::SetWorldGroupInfo(
    std::unordered_map<std::string, std::map<u32, HcclIpAddress>> phyIdNicInfoMap,
    vector<RankInfo> worldRankInfoList, vector<u32> ranksPort)
{
    for (auto &ipInfo : phyIdNicInfoMap) {
        for (auto &devInfo : ipInfo.second) {
            rankDevicePhyIdNicInfoMap_[ipInfo.first][devInfo.first] = devInfo.second;
            HCCL_INFO("phyIdNicInfoMap print hostIp[%s] devId[%u] devIp[%s]",
                ipInfo.first.c_str(), devInfo.first, devInfo.second.GetReadableAddress());
        }
    }

    for (auto &rankInfo : worldRankInfoList) {
        worldRankInfoList_.push_back(rankInfo);
    }

    for (auto &rank : ranksPort) {
        ranksPort_.push_back(rank);
        HCCL_INFO("ranksPort port[%u]", rank);
    }
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::GetTopoDesc(HcclTopoDescs *topoDescs, uint32_t topoSize)
{
    if (topoSize < static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_MAX)) {
        HCCL_ERROR("topoDescs size is not enough, please check topoSize[%u]", topoSize);
        return HCCL_E_PARA;
    }

    if (deviceType_ == DevType::DEV_TYPE_910_93) {
        topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].algSets = HCCL_ALG_SWITCH | HCCL_ALG_RING;
        topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].algSets = HCCL_ALG_RING;
    } else if (deviceType_ == DevType::DEV_TYPE_910B) {
        topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].algSets = HCCL_ALG_MESH;
        topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].algSets = 0;
    } else if (deviceType_ == DevType::DEV_TYPE_310P3) {
        topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].algSets = HCCL_ALG_RING;
        topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].algSets = 0;
    }

    topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].rankSize = userRankSize_;
    topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].rankSize = 0;
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::NetWorkRdmaSetQpDepth()
{
    RaResourceInfo raResourceInfo;
    void *nicRdmaHandle = nullptr;
    u32 qpDepth = 0;
    u32 qpNum = 0;
    CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
    auto it = raResourceInfo.nicSocketMap.find(devIpAddr_[0]);
    if (it == raResourceInfo.nicSocketMap.end()) {
        HCCL_ERROR("[GetRaResourceInfo]nic socket handle did not found");
        return HCCL_E_PARA;
    }
    nicRdmaHandle = it->second.nicRdmaHandle;
    CHK_PTR_NULL(nicRdmaHandle);

    u32 esMaxPsTable = GetEsMaxPsTable();
    u32 esStreamNum = GetEsStreamNum();
    if (esStreamNum == 0) {
        HCCL_ERROR("esStreamNum[%u] cannot be 0", esStreamNum);
        return HCCL_E_PARA;
    }

    // 通过GE传入的tag进行计算合理的QP_Depth, 找到向上最近大于自己的2的幂次方
    u32 esMaxRemoteopNumPerStream = static_cast<u32>(ceil(static_cast<double>(esMaxPsTable) / esStreamNum));
    u32 optimalQpDepth = esMaxRemoteopNumPerStream * 2;
    HCCL_INFO("esMaxRemoteopNumPerStream[%u], optimalQpDepth[%u]",
        esMaxRemoteopNumPerStream, optimalQpDepth);
    if (optimalQpDepth > MAX_QP_DEPTH_HCCP) {
        HCCL_ERROR("optimalQpDepth[%u] cannot exceed %u", optimalQpDepth, MAX_QP_DEPTH_HCCP);
        return HCCL_E_PARA;
    } else if (optimalQpDepth == 0) {
        optimalQpDepth = MAX_QP_DEPTH;
    } else {
        optimalQpDepth = optimalQpDepth < MIN_QP_DEPTH ? MIN_QP_DEPTH : optimalQpDepth;
    }

    HCCL_INFO("Final optimalQpDepth[%u]", optimalQpDepth);
    CHK_RET(HrtRaGetQpDepth(nicRdmaHandle, &qpDepth, &qpNum));
    HCCL_INFO("GetQpDepth: qpDepth %u qpNum %u", qpDepth, qpNum);
    if (qpDepth != optimalQpDepth) {
        CHK_RET(HrtRaSetQpDepth(nicRdmaHandle, optimalQpDepth, &qpNum));
        HCCL_INFO("SetQpDepth: qp depth %u qpNUm %u", optimalQpDepth, qpNum);
    }

    return  HCCL_SUCCESS;
}

HcclResult HcclImplBase::CreateTagAndBuildLinks(int32_t tag, uint32_t workerRanks[], uint64_t workerNum)
{
    HCCL_INFO("CreateTagAndBuildLinks tag[%u]", tag);
    if (workerNum == 0) {
        HCCL_WARNING("[CreateTagAndBuildLinks]workerNum is zero Default build all worker rank links");
        isEsAllLinkModel_ = true;
        CHK_RET(BuildEsPsTransportLinks(tag, false));
        return HCCL_SUCCESS;
    }

    std::unique_lock<std::mutex> tagNodesInfoLock(tagNodesInfoLock_);
    if (tagNodesInfo_.find(tag) != tagNodesInfo_.end()) {
        HCCL_ERROR("[CreateTagAndBuildLinks]tag[%d] has been created, please check", tag);
        return HCCL_E_PARA;
    }

    vector<u32> clientRanksList(workerRanks, workerRanks + workerNum);
    if (unordered_set<int>(clientRanksList.begin(), clientRanksList.end()).size() != clientRanksList.size()) {
        HCCL_ERROR("[CreateTagAndBuildLinks]tag[%d] workerRanks can't contains duplicate elements", tag);
        return HCCL_E_PARA;
    }

    u32 nodesInfoNum = 0;
    for (auto rankId : clientRanksList) {
        for (auto &nodeInfo : clientsInfo_) {
            if (rankId == nodeInfo.rankId) {
                tagNodesInfo_[tag].push_back(nodeInfo);
                break;
            }
        }
        if (tagNodesInfo_[tag].size() > nodesInfoNum) {
            nodesInfoNum++;
        } else {
            HCCL_ERROR("[CreateTagAndBuildLinks]tag[%d] workerRanks[%u] can't find from clientsInfo_", tag, rankId);
            return HCCL_E_PARA;
        }
    }

    for (auto iter : tagNodesInfo_[tag]) {
        HCCL_INFO("[CreateTagAndBuildLinks]tagNodesInfo_ tag[%d] remoteRank[%u]", tag, iter.rankId);
    }
    tagNodesInfoLock.unlock();

    CHK_RET(BuildEsPsTransportLinks(tag, true));
    return HCCL_SUCCESS;
}

HcclResult HcclImplBase::BuildEsPsTransportLinks(const s32 &tag, bool isTagAssociate)
{
    uint64_t beginTime = hrtMsprofSysCycleTime();
    for (const auto &node : serversInfo_) {
        if (node.rankId == userRank_) {
            HCCL_DEBUG("buildTransportRefCount_[%u] firstBuildTransportFlag_[%d]",
                buildTransportRefCount_, firstBuildTransportFlag_);
            std::unique_lock<std::mutex> lock(socketListenMutex_);
            if (buildTransportRefCount_ == 0) {
                if (!firstBuildTransportFlag_) {
                    CHK_RET(NetworkManager::GetInstance(deviceLogicId_).HeterogStartListen(loopBackIp_,
                        ranksPort_[userRank_]));
                } else {
                    firstBuildTransportFlag_ = false;
                }
            }
            buildTransportRefCount_++;
            lock.unlock();

            std::unique_lock<std::mutex> tagNodesInfoLock(tagNodesInfoLock_);
            std::vector<RoleTableNodeInfo> &nodesInfo = isTagAssociate == true ? tagNodesInfo_[tag] : clientsInfo_;
            tagNodesInfoLock.unlock();
            CHK_RET(BuildTransport(node.port, node.hostPort, tag, nodesInfo));

            lock.lock();
            buildTransportRefCount_--;
            if (buildTransportRefCount_ == 0) {
                CHK_RET(NetworkManager::GetInstance(deviceLogicId_).HeterogStopListen(loopBackIp_, ranksPort_[userRank_]));
            }
            break;
        }
    }

    auto &profilingManager = ProfilingManager::Instance();
    CHK_RET(profilingManager.CallEsMsprofReportTaskApi(false, beginTime, ProfTaskType::TASK_BUILD_CS_TRANSPORT));
    return HCCL_SUCCESS;
}
}
